Netty实战与源码剖析（二）——基于NIO的群聊系统

以你之姓@ 2022-09-12 12:58 194阅读 0赞

## 服务端实现 ##

package com.Zhongger.groupchat;
    
    import java.io.IOException;
    import java.net.InetSocketAddress;
    import java.nio.ByteBuffer;
    import java.nio.channels.*;
    import java.util.Iterator;
    import java.util.Set;
    
    /** * @author zhongmingyi * @date 2021/9/25 3:07 下午 */
    public class GroupChatServer { 
        private Selector selector;
        private ServerSocketChannel serverSocketChannel;
        private static final int PORT = 6666;
    
        public GroupChatServer() { 
            try { 
                selector = Selector.open();
                serverSocketChannel = ServerSocketChannel.open();
                serverSocketChannel.configureBlocking(false);
                serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
                serverSocketChannel.bind(new InetSocketAddress(PORT));
            } catch (Exception e) { 
                e.printStackTrace();
            }
        }
    
        //监听
        public void listen() { 
            try { 
                while (true) { 
                    int count = selector.select();
                    if (count > 0) { 
                        //有事件处理
                        Set<SelectionKey> selectionKeys = selector.selectedKeys();
                        Iterator<SelectionKey> iterator = selectionKeys.iterator();
                        while (iterator.hasNext()) { 
                            SelectionKey selectionKey = iterator.next();
                            if (selectionKey.isAcceptable()) { 
                                SocketChannel socketChannel = serverSocketChannel.accept();
                                socketChannel.configureBlocking(false);
                                socketChannel.register(selector, SelectionKey.OP_READ);
                                System.out.println(socketChannel.getRemoteAddress() + "上线了");
                            }
                            if (selectionKey.isReadable()) { 
                                //处理读
                                readFromClient(selectionKey);
                            }
                            //手动将当前SelectionKey移除，防止重复处理
                            iterator.remove();
                        }
                    } else { 
                        System.out.println("等待事件......");
                    }
                }
            } catch (Exception e) { 
                e.printStackTrace();
            } finally { 
    
            }
        }
    
        private void readFromClient(SelectionKey selectionKey) { 
            SocketChannel channel = null;
            try { 
                channel = (SocketChannel) selectionKey.channel();
                ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
                int read = channel.read(byteBuffer);
                if (read > 0) { 
                    String msg = new String(byteBuffer.array());
                    System.out.println("from 客户端：" + msg);
                    //转发消息给其他客户端
                    sendInfoToOtherClient(msg, channel);
                }
            } catch (Exception e) { 
                try { 
                    System.out.println(channel.getRemoteAddress() + "离线了...");
                    //取消注册
                    selectionKey.cancel();
                    //关闭通道
                    channel.close();
                } catch (IOException ioException) { 
                    ioException.printStackTrace();
                }
            }
        }
    
    
        private void sendInfoToOtherClient(String msg, SocketChannel selfSocketChannel) throws IOException { 
            System.out.println("服务器转发消息中...");
            //遍历所有注册到selector的key，并排查自己
            for (SelectionKey selectionKey : selector.keys()) { 
                Channel targetChannel = selectionKey.channel();
                //排除自己
                if (targetChannel instanceof SocketChannel && targetChannel != selfSocketChannel) { 
                    SocketChannel destChannel = (SocketChannel) targetChannel;
                    destChannel.write(ByteBuffer.wrap(msg.getBytes()));
                }
            }
        }
    
        public static void main(String[] args) { 
            GroupChatServer server = new GroupChatServer();
            server.listen();
        }
    }

## 客户端实现 ##

package com.Zhongger.groupchat;
    
    import java.io.IOException;
    import java.net.InetSocketAddress;
    import java.nio.ByteBuffer;
    import java.nio.channels.SelectionKey;
    import java.nio.channels.Selector;
    import java.nio.channels.SocketChannel;
    import java.util.Iterator;
    import java.util.Scanner;
    import java.util.Set;
    
    /** * @author zhongmingyi * @date 2021/9/25 3:44 下午 */
    public class GroupChatClient { 
        private final String HOST = "127.0.0.1";
        private final Integer PORT = 6666;
        private Selector selector;
        private SocketChannel socketChannel;
        private String username;
    
        public GroupChatClient() throws IOException { 
            selector = Selector.open();
            socketChannel = SocketChannel.open(new InetSocketAddress(HOST, PORT));
            socketChannel.configureBlocking(false);
            socketChannel.register(selector, SelectionKey.OP_READ);
            username = socketChannel.getLocalAddress().toString().substring(1);
            System.out.println(username + " is OK ...");
        }
    
        public void sendInfo(String info) { 
            info = username + "说：" + info;
            try { 
                socketChannel.write(ByteBuffer.wrap(info.getBytes()));
            } catch (Exception e) { 
                e.printStackTrace();
            }
        }
    
        public void readInfo() { 
            try { 
                int readChannels = selector.select();
                if (readChannels > 0) { 
                    Set<SelectionKey> selectionKeys = selector.selectedKeys();
                    Iterator<SelectionKey> iterator = selectionKeys.iterator();
                    while (iterator.hasNext()) { 
                        SelectionKey selectionKey = iterator.next();
                        if (selectionKey.isReadable()) { 
                            SocketChannel channel = (SocketChannel) selectionKey.channel();
                            ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
                            channel.read(byteBuffer);
                            String msg = new String(byteBuffer.array());
                            System.out.println(msg.trim());
                        }
                        iterator.remove();
                    }
                } else { 
                    System.out.println("无可用的通道...");
                }
            } catch (Exception e) { 
                e.printStackTrace();
            }
        }
    
        public static void main(String[] args) throws IOException { 
            GroupChatClient client = new GroupChatClient();
            new Thread(() -> { 
                while (true) { 
                    client.readInfo();
                    try { 
                        Thread.sleep(3000);
                    } catch (Exception e) { 
                        e.printStackTrace();
                    }
                }
            }).start();
    
            //发送数据
            Scanner scanner = new Scanner(System.in);
            while (scanner.hasNextLine()) { 
                String msg = scanner.nextLine();
                client.sendInfo(msg);
            }
        }
    }