Netty#

1
public class GroupChatServer {
2
    // 定义属性
3
    private Selector selector;
4
    private ServerSocketChannel listenChannel;
5
    private static final int PORT = 6667;
6

7
    // 构造器
8
    // 初始化工作
9
    public GroupChatServer() {
10
        try {
11
            // 得到选择器
12
            selector = Selector.open();
13
            // ServerSocketChannel
14
            listenChannel = ServerSocketChannel.open();
15
            // 绑定端口
16
            listenChannel.socket().bind(new InetSocketAddress(PORT));
17
            // 设置非阻塞模式
18
            listenChannel.configureBlocking(false);
19
            // 将该listenChannel 注册到 Selector
20
            listenChannel.register(selector, SelectionKey.OP_ACCEPT);
21
        } catch (IOException e) {
22
            e.printStackTrace();
23
        }
24
    }
25

26
    // 监听
27
    public void listen() {
28
        try {
29
            // 循环处理
30
            while (true) {
31
                //这个方法会阻塞，直到至少有一个通道准备好进行I/O操作（如读、写、连接等）。
32
                //如果有通道准备好，select() 方法会返回准备好的通道数量。
33
                int count = selector.select();
34
                //如果 count 大于 0，说明有通道准备好进行I/O操作。
35
                //代码会遍历 selector.selectedKeys() 返回的 SelectionKey 集合，处理每个 SelectionKey。
36
                if (count > 0) { // 有事件处理
37
                    // 遍历得到的selectedkey集合
38
                    Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
39
                    while (iterator.hasNext()) {
40
                        SelectionKey key = iterator.next();
41
                        //如果 key.isAcceptable() 为 true，说明有新的连接请求，代码会接受连接并将其注册到 Selector 中。
42
                        if (key.isAcceptable()) {// 监听到sccept
43
                            SocketChannel sc = listenChannel.accept();
44
                            sc.configureBlocking(false);
45
                            // 将该scoketChannel注册到Selector中
46
                            sc.register(selector, SelectionKey.OP_READ);
47
                            // 提示
48
                            System.out.println(sc.getRemoteAddress() + " 已上线.....");
49
                        }
50
                        //如果 key.isReadable() 为 true，说明通道有数据可读，代码会调用 readData(key) 方法处理读取的数据。
51
                        if (key.isReadable()) {// 通道发生read事件，即通道是可读状态
52
                            // 处理读
53
                            readData(key);
54
                        }
55
                        //处理完一个 SelectionKey 后，需要将其从集合中移除，以防止重复处理。
56
                        iterator.remove();
57
                    }
58
                } else {
59
                    //如果 count 为 0，说明当前没有通道准备好进行I/O操作，代码会输出 "等待中......"。
60
                    System.out.println("等待中......");
61
                }
62
            }
63
        } catch (Exception e) {
64
            e.printStackTrace();
65
        } finally {
66

67
        }
68
    }
69

70
    // 读取客户端消息
71
    private void readData(SelectionKey key) {
72
        // 定义一个SocketChannel
73
        SocketChannel channel = null;
74
        try {
75
            // 取得关联的channel
76
            channel = (SocketChannel) key.channel();
77
            // 创建buffer
78
            ByteBuffer buffer = ByteBuffer.allocate(1024);
79

80
            int count = channel.read(buffer);
81
            // 根据count的值做处理
82
            if (count > 0) {
83
                // 把缓冲区的数据转成字符串
84
                String msg = new String(buffer.array());
85
                // 输出该消息
86
                System.out.println("from 客户端： " + msg.trim());
87

88
                // 向其他的客户端转发消息（排除自己）
89
                sendInfoToOtherClients(msg, channel);
90
            }
91
        } catch (IOException e) {
92
            try {
93
                System.out.println(channel.getRemoteAddress() +  "离线.......");
94
                // 取消注册
95
                key.cancel();
96
                // 关闭通道
97
                channel.close();
98
            } catch (IOException ex) {
99
                ex.printStackTrace();
100
            }
101
        }
102

103
    }
104
    // 转发消息给其他客户（通道）
105
    private void sendInfoToOtherClients(String msg, SocketChannel self) throws IOException {
106
        System.out.println("服务器转发消息中.......");
107
        // 遍历 所有注册到selector 上的SocketChannel，并排除自己
108
        for (SelectionKey key : selector.keys()) {
109
            // 通过 key 取出对应的 SocketChannel
110
            Channel targetChannel = key.channel();
111
            // 排除自己
112
            if(targetChannel instanceof SocketChannel && targetChannel != self){
113
                // 转型
114
                SocketChannel dest =  (SocketChannel)targetChannel;
115
                // 将msg 存储到buffer
116
                ByteBuffer buffer = ByteBuffer.wrap(msg.getBytes());
117
                // 将buffer数据写入到通道中
118
                dest.write(buffer);
119
            }
120
        }
121
    }
122

123
    public static void main(String[] args) {
124
        // 创建服务器对象
125
        GroupChatServer groupChatServer = new GroupChatServer();
126
        // 监听
127
        groupChatServer.listen();
128
    }
129
}

1
/**
2
 * 客户端代码
3
 */
4
public class GroupChatClient {
5
    // 定义相关的属性
6
    // 服务器IP
7
    private final String HOST = "127.0.0.1";
8
    // 服务器端口
9
    private final int PORT = 6667;
10
    private Selector selector;
11
    private SocketChannel socketChannel;
12
    private String username;
13

14
    // 构造器
15
    public GroupChatClient() throws IOException {
16
        selector = Selector.open();
17
        // 连接服务器
18
        socketChannel = SocketChannel.open(new InetSocketAddress(HOST, PORT));
19
        // 设置非阻塞
20
        socketChannel.configureBlocking(false);
21
        // 将channel 注册到selector
22
        socketChannel.register(selector, SelectionKey.OP_READ);
23
        // 得到username
24
        username = socketChannel.getLocalAddress().toString().substring(1);
25
        System.out.println(username + " is ok......");
26
    }
27

28
    // 向服务器发送消息
29
    public void senInfo(String info) {
30
        info = username + " 说：" + info;
31
        try {
32
            socketChannel.write(ByteBuffer.wrap(info.getBytes()));
33
        } catch (IOException e) {
34
            e.printStackTrace();
35
        }
36
    }
37

38
    // 读取从服务器端回复的消息
39
    public void readInfo() {
40
        try {
41
            int readChannel = selector.select();
42
            if (readChannel > 0) {// 即有可以用通道
43
                Set<SelectionKey> selectionKeys = selector.selectedKeys();
44
                Iterator<SelectionKey> iterator = selectionKeys.iterator();
45
                while (iterator.hasNext()) {
46
                    SelectionKey key = iterator.next();
47
                    if (key.isReadable()) {
48
                        SocketChannel channel = (SocketChannel) key.channel();
49
                        // 得到buffer
50
                        ByteBuffer buffer = ByteBuffer.allocate(1024);
51
                        // 读取
52
                        channel.read(buffer);
53
                        // 把读到的缓冲区的数据转成字符串
54
                        String msg = new String(buffer.array());
55
                        System.out.println(msg.trim());
56
                    }
57
                }
58
                // 删除当前的SelectionKey，防止重复操作
59
                iterator.remove();
60
            } else {
61
//                System.out.println("没有可用的通道.....");
62

63
            }
64
        } catch (Exception e) {
65
            e.printStackTrace();
66
        }
67
    }
68
    public static void main(String[] args) throws IOException {
69
        // 启动客户端
70
        GroupChatClient chatClient = new GroupChatClient();
71

72
        // 启动一个线程，每隔3秒读取从服务器端发送的数据
73
        new Thread(){
74
            @Override
75
            public void run() {
76
                while (true){
77
                    chatClient.readInfo();
78
                    try {
79
                        Thread.currentThread().sleep(3000);
80
                    }catch (InterruptedException e){
81
                        e.printStackTrace();
82
                    }
83
                }
84
            }
85
        }.start();
86

87
        // 发送数据给服务器端
88
        Scanner scanner = new Scanner(System.in);
89
        while (scanner.hasNextLine()){
90
            String str = scanner.next();
91
            chatClient.senInfo(str);
92
        }
93

94
    }
95
}