NIO 简易版群聊系统

﹏ヽ暗。殇╰゛Y 2022-09-02 12:55 12阅读 0赞

服务端启动后，接收客户端发送的消息，把当前消息转发到选择器中注册的其他客户端对应的通道中，实现简单群聊系统。

服务端：

package com.tech.netty.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 lw
     * @since 2021/8/3
     **/
    public class GroupChatServer {
    
        private Selector selector;
        private ServerSocketChannel serverSocketChannel;
        private static final int PORT=6667;
    
        //构造器
        public GroupChatServer() {
            try {
                //创建选择器
                selector=Selector.open();
                //创建ServerSocketChannel
                serverSocketChannel = ServerSocketChannel.open();
                //绑定端口
                serverSocketChannel.bind(new InetSocketAddress(PORT));
                //设置非阻塞
                serverSocketChannel.configureBlocking(false);
                //在选择器上注册 客户端连接事件
                serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    
        //监听请求并处理
        public void listen(){
            try {
                //循环处理
                while(true){
                    int selectCount = selector.select();
                    //有事件处理
                    if(selectCount>0){
                        //遍历得到SelectionKey集合
                        Set<SelectionKey> selectionKeys = selector.selectedKeys();
                        Iterator<SelectionKey> iterator = selectionKeys.iterator();
                        while(iterator.hasNext()){
                            //取出SelectionKey
                            SelectionKey selectionKey = iterator.next();
                            //客户端连接事件
                            if(selectionKey.isAcceptable()){
                                SocketChannel socketChannel = serverSocketChannel.accept();
                                socketChannel.configureBlocking(false);
                                //SocketChannel在选择器上注册数据读取事件
                                socketChannel.register(selector,SelectionKey.OP_READ);
                                //提升上线
                                System.out.println(socketChannel.getRemoteAddress()+"上线");
                            }
                            //判断为数据读取事件
                            if (selectionKey.isReadable()){
                                readData(selectionKey);
                            }
                            iterator.remove();
                        }
    
                    }else {
                        System.out.println("等待...");
                    }
    
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    
        private void readData(SelectionKey selectionKey) {
            SocketChannel socketChannel = null;
            try {
                socketChannel = (SocketChannel) selectionKey.channel();
                ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
                //通道数据读取到缓冲区
                int readCount = socketChannel.read(byteBuffer);
                if(readCount>0){
                    String msg=new String(byteBuffer.array());
                    System.out.println("from 客户端:"+msg);
                    //将msg发送到选择器上其他客户端的通道（除去当前通道）
                    sendInfoToOtherClients(socketChannel,msg);
                }
            } catch (IOException e) {
                try {
                    System.out.println(socketChannel.getRemoteAddress()+"离线了");
                    //取消注册
                    selectionKey.cancel();
                    //关闭通道
                    socketChannel.close();
                } catch (IOException e1) {
                    e1.printStackTrace();
                }
            }
        }
    
        private void sendInfoToOtherClients(SocketChannel socketChannel, String msg) throws IOException {
            System.out.println("转发消息中...");
    
            //遍历所有在selector上注册的channel 找出非当前SocketChannel给客户端转发消息
            for(SelectionKey selectionKey:selector.keys()){
                Channel channel = selectionKey.channel();
                if(channel instanceof SocketChannel && channel!=socketChannel){
                    ByteBuffer byteBuffer = ByteBuffer.wrap(msg.getBytes());
                    ((SocketChannel) channel).write(byteBuffer);
                }
            }
        }
    
        public static void main(String[] args) {
            GroupChatServer groupChatServer = new GroupChatServer();
            groupChatServer.listen();
        }
    }

客户端：

package com.tech.netty.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 lw
     * @since 2021/8/3
     **/
    public class GroupChatClient {
    
        private final int PORT = 6667;
        private final String HOST = "127.0.0.1";
        private SocketChannel socketChannel;
        private Selector selector;
        private String userName;
    
        public GroupChatClient() throws IOException {
            selector = Selector.open();
            //连接服务器
            socketChannel = SocketChannel.open(new InetSocketAddress(HOST, PORT));
            //设置非阻塞
            socketChannel.configureBlocking(false);
            //channel在选择器上注册数据读取事件
            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 selectCount = selector.select();
                //有可以用的通道
                if(selectCount>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());
                        }
                        //从集合中删除SelectionKey 防止重复操作
                        iterator.remove();
                    }
                }
    
            }catch (Exception e){
    
            }
        }
    
        public static void main(String[] args) throws IOException {
            //启动客户端
            GroupChatClient groupChatClient = new GroupChatClient();
    
            //每隔3秒读取从服务器发送的数据
            new Thread(()->{
                while (true){
                    groupChatClient.readInfo();
                    try {
                        Thread.sleep(3000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }).start();
    
            //发送数据给服务端
            Scanner scanner = new Scanner(System.in);
            while (scanner.hasNextLine()){
                String s = scanner.nextLine();
                groupChatClient.sendInfo(s);
            }
        }
    }