Java NIO实现聊天系统

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
/**
 * 群聊的服端
 *
 * @author :breakpoint/赵立刚
 * @date : 2020/08/13
 */
public class GroupChatServer {

    // 定义相关的属性
    private Selector selector;
    private ServerSocketChannel listenChannel;
    private static final int port = 6667;

    // 构造器
    // 进行初始化的操作
    public GroupChatServer() {
        try {
            // 获取选择器
            selector = Selector.open();
            // 获取到 listenChannel
            listenChannel = ServerSocketChannel.open();
            // 设定端口
            listenChannel.bind(new InetSocketAddress(port));
            // 设定非阻塞模式
            listenChannel.configureBlocking(false);
            // 将该 listenChannel 注册到 selector上 完成操作
            listenChannel.register(selector, SelectionKey.OP_ACCEPT);
            System.out.println("服务器启动了。。。。");
        } catch (IOException e) {

        }

    }

    // 监听的代码

    public void listen() {
        try {
            // 循环处理
            while (true) {
                int count = selector.select(2000);
                if (count > 0) {
                    // 有事件需要处理
                    // 遍历处理 得到selectionKeys集合
                    Set<SelectionKey> selectionKeys = selector.selectedKeys();
                    Iterator<SelectionKey> selectionKeyIterator = selectionKeys.iterator();
                    while (selectionKeyIterator.hasNext()) {
                        // 得到selectionKey
                        SelectionKey selectionKey = selectionKeyIterator.next();
                        // 监听到了 accept
                        if (selectionKey.isAcceptable()) {
                            // 获取到连接
                            SocketChannel sc = listenChannel.accept();
                            sc.configureBlocking(false);
                            sc.register(selector, SelectionKey.OP_READ, ByteBuffer.allocate(1024));
                            // 提示上线
                            System.out.println(sc.getRemoteAddress() + ":上线啦。。。。");
                        }

                        if (selectionKey.isReadable()) {
                            // 读取事件 通道是可以读的状态 //专门写
                            readData(selectionKey);
                        }


                        // 移除当前的删除 防止重复处理操作
                        selectionKeyIterator.remove();
                    }


                } else {
                    System.out.println("等待中。。。。。。");
                }
            }

        } catch (Exception e) {

        } finally {

        }
    }

    // 读取客户端的消息

    private void readData(SelectionKey selectionKey) {
        // 获取 socketChannel

        SocketChannel channel = null;
        try {
            channel = (SocketChannel) selectionKey.channel();
            ByteBuffer byteBuffer = (ByteBuffer) selectionKey.attachment();
            int count = channel.read(byteBuffer);
            // 分情况处理
            if (count > 0) {
                // 获取到数据专程
                String msg = new String(byteBuffer.array(), 0, count);
                byteBuffer.clear();
                System.out.println(channel.getRemoteAddress() + "来自客户端" + msg);
                // 向其他的客户端转发消息
                sendInfoToOtherClients(msg, channel);
            }

        } catch (IOException e) {
            // 如果发生异常 提示说明离线了
            try {
                System.out.println(channel.getRemoteAddress() + "离线了。。。。");
                // 取消注册
                selectionKey.cancel();
                // 关闭通道
                channel.close();
            } catch (IOException e1) {
                //e1.printStackTrace();
            }
        } finally {

        }
    }


    // 转发消息给其他的客户端 去掉自己的客户端
    private void sendInfoToOtherClients(String msg, SocketChannel self) throws IOException {
        System.out.println("服务器转发消息。。。。。");
        // 进行遍历操作
        Set<SelectionKey> keys = selector.keys();
        for (SelectionKey key : keys) {
            // 取出来所有的
            Channel targetChannel = key.channel();
            // 排除自己
            if (targetChannel instanceof SocketChannel && targetChannel != self) {
                SocketChannel dest = (SocketChannel) targetChannel;
                ByteBuffer byteBuffer = ByteBuffer.wrap(msg.getBytes());
                // 发送数据
                dest.write(byteBuffer);
            }
        }
    }


    public static void main(String[] args) {
        GroupChatServer groupChatServer = new GroupChatServer();
        groupChatServer.listen();
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
/**
 * @author :breakpoint/赵立刚
 * @date : 2020/08/13
 */
public class GroupChatClient {

    // 定义相关属性

    private final String HOST = "127.0.0.1"; //服务器地址
    private final int port = 6667; // 服务器端口
    private Selector selector;
    private SocketChannel socketChannel;
    private String userName;

    // 完成初始化工作
    public GroupChatClient() {
        try {
            selector = Selector.open();
            // 连接服务器
            socketChannel = SocketChannel.open(new InetSocketAddress(HOST, port));
            // 设置非阻塞工作
            socketChannel.configureBlocking(false);
            // 注册我们的通道
            socketChannel.register(selector, SelectionKey.OP_READ);
            userName = socketChannel.getLocalAddress().toString();
            System.out.println("客户端专备好啦");
        } catch (IOException e) {

        }
    }

    public void sendInfo(String info) {
        String msg = userName + "说：" + info;
        try {
            ByteBuffer wrap = ByteBuffer.wrap(msg.getBytes());
            socketChannel.write(wrap);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    // 读取信息
    public void readInfo() {
        try {
            int readChannel = selector.select(2000);
            if (readChannel > 0) {
                // 有可以用的通道
                Set<SelectionKey> selectionKeys = selector.selectedKeys();
                Iterator<SelectionKey> iterator = selectionKeys.iterator();
                while (iterator.hasNext()) {
                    SelectionKey next = iterator.next();
                    if (next.isReadable()) {
                        SelectableChannel keyChannel = next.channel();
                        if (keyChannel instanceof SocketChannel) {
                            // 获取到我们的通道
                            SocketChannel channel = (SocketChannel) keyChannel;
                            ByteBuffer allocate = ByteBuffer.allocate(1024);
                            // 读取数据
                            int read = channel.read(allocate);
                            if (read > 0) {
                                // 输出我们的消息
                                System.out.println(new String(allocate.array(), 0, read));
                            }

                        }// end if
                    }
                    iterator.remove();
                }
            } else {
                System.out.println("没有可用的通道");
            }
        } catch (IOException e) {

        }
    }

    public static void main(String[] args) throws Exception {


        // 启动客户端的操作
        final GroupChatClient groupChatClient = new GroupChatClient();


        // 启动一个线程

        new Thread(() -> {
            while (true) {
                groupChatClient.readInfo();
                try {
                    Thread.currentThread().sleep(3000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();


        Scanner scanner = new Scanner(System.in);

        while (scanner.hasNext()) {
            // 输入信息
            String s = scanner.nextLine();
            groupChatClient.sendInfo(s);
        }


        System.in.read();
    }
}