(六) socket·NIO socket 实现控制台一问一答式聊天

女爷i 2022-04-23 10:30 229阅读 0赞

### 文章目录 ###

*   *   *   *   *  前言
                 *  代码
                 *   *  NIO socket 服务端
                     *  NIO socket 客户端
                 *  控制台信息
                 *   *  服务端
                     *  客户端

##### 前言 #####

> 基于[(五) socket·NIO socket 的基本介绍][socket_NIO socket]中对 NIO socket 的介绍，本文实现在控制台一问一答式的聊天。

##### 代码 #####

> 控制台打印过程将体现服务端和客户端采取的措施。  
> 代码逻辑可以看注释。  
> 如果使用多线程的话，服务端和客户端可以不限制次序的聊天，本文没有使用多线程，便于理解 NIO socket 的交互流程  
> 服务器端和客户端都是通过运行 main 方法启动

###### NIO socket 服务端 ######

> NIO socket 服务端支持非 NIO socket 服务的请求

import java.io.IOException;
    import java.io.UnsupportedEncodingException;
    import java.net.InetSocketAddress;
    import java.nio.ByteBuffer;
    import java.nio.channels.*;
    import java.nio.channels.spi.SelectorProvider;
    import java.util.*;
    
    /** * @Title: Socket Server * @Description: NIO Socket * @Author: WuJie */
    public class NioSocketServer { 
        private final static String CHARSET="UTF8";//设置编码格式
        private static Selector selector=null;//nio Selector
    
        public static void main(String [] args){ 
            NioSocketServer nioSocketServer=new NioSocketServer();
            nioSocketServer.startServer();//启动服务器，等待客户端访问
            if(selector!=null){ 
                nioSocketServer.handle(selector);//处理客户端的请求
            }else{ 
                System.out.println("error 2,服务端启动失败");
            }
        }
    
        /** * 启动服务器 */
        public void startServer(){ 
    
            try{ 
                selector=SelectorProvider.provider().openSelector();//初始化 nio Selector
                selector.select(1000);//阻塞 selector 1000 毫秒，非必须
    
                //Channel 多线程安全、支持异步关闭
                ServerSocketChannel serverSocketChannel=ServerSocketChannel.open();//初始化服务端 SocketChannel
                serverSocketChannel.bind(new InetSocketAddress(10000));//设置服务端端口
                serverSocketChannel.configureBlocking(false);//非阻塞模式，否则就不是NIO了
    
                serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);//将服务端 Channel 注册到 Selector,感兴趣事件为 准备接受访问
                System.out.println("服务端已启动,服务器端字符编码格式为"+CHARSET+",等待客户端的访问：");
            }catch(IOException e){ 
                System.out.println("error 1,获取 Selector 失败"+e);
            }
        }
    
        /** * 处理客户端的请求与交互 * @param selector */
        public void handle(Selector selector){ 
            try{ 
                for(;;){ //服务端始终处于启动状态
                    int keyNumbers=selector.select();//获取已注册 channel 数量
                    if(keyNumbers>0){ //如果已注册 channel 数量>0,则开始处理
                        Set<SelectionKey> selectionKeySet=selector.selectedKeys();//获取 selector 中已注册 channel 的 SelectionKey 集合
    
                        Iterator iterable=selectionKeySet.iterator();//准备遍历 selectionKeySet，处理其对应的 channel
                        while(iterable.hasNext()){ 
                            SelectionKey selectionKey=(SelectionKey)iterable.next();
                            iterable.remove();//Removes from the underlying collection the last element returned by this iterator (optional operation)
    
                            if(selectionKey.isAcceptable()){ //接受客户端请求
                                acceptable(selectionKey);//TODO 可以使用多线程
                            }else if(selectionKey.isReadable()&&selectionKey.isValid()){ //读取客户端传输的数据
                                readable(selectionKey);//TODO 可以使用多线程
                            }else if(selectionKey.isWritable()&&selectionKey.isValid()) { //返回给客户端信息
                                writable(selectionKey);//TODO 可以使用多线程
                            }
    
                        }
                    }else{ 
                        continue;
                    }
                }
    
            }catch(IOException e){ 
                System.out.println("error 3,获取 Selector 已注册 Channel 数量失败"+e);
            }
    
        }
    
        /** * 处理客户端请求 * 将客户单 channel 注册到 selector * @param selectionKey */
        private static void acceptable(SelectionKey selectionKey){ 
            try{ 
                System.out.println("可接收--------------开始");
    
                //获取客户端请求
                SocketChannel clientChannel=((ServerSocketChannel) selectionKey.channel()).accept();
                clientChannel.configureBlocking(false);
    
                Map map=new HashMap<String,String>();//可以将该map连同 channel 一起注册到 selector,map可以携带附属信息
                map.put("demo","没啥用，就是为了表示可以携带一些额外信息，selectionKey 可以获取");
    
                clientChannel.register(selector, SelectionKey.OP_READ,map);
                System.out.println("【selectionKey.isAcceptable()=true，从select 获取 服务端channel ，从这个channel 获取客户端 channel，向 selector 注册 客户端 channel,感兴趣事件为 OP_READ】");
                System.out.println("可接收--------------结束");
                System.out.println();
            }catch(ClosedChannelException e1){ 
                System.out.println("error 4,将客户端请求 channel 注册到 selector 失败"+e1);
            }catch (IOException e2){ 
                System.out.println("error 5,获取客户端请求失败"+e2);
            }
        }
    
        /** * 处理可读的channel * 读取客户端发送的信息 * @param selectionKey */
        private static void readable(SelectionKey selectionKey){ 
            try{ 
                System.out.println("可读--------------开始");
    
                SocketChannel clientChannel=(SocketChannel) selectionKey.channel();//根据key获取channel
                ByteBuffer byteBuffer= ByteBuffer.allocate(1024); //准备 1k 的缓冲区
                int len=clientChannel.read(byteBuffer);//将channel 内容写入缓存 byteBuffer
                //System.out.println("byteBuffer.remaining()="+byteBuffer.remaining()+";len="+len);
    
                if(len==-1){ //如果没有信息传入，说明客户端可能中断连接，则关闭这个客户端channel
                    try{ 
                        clientChannel.close();
                        selector.wakeup();//唤起其他channel
                        System.out.println("line 133,客户端已关闭:"+clientChannel.socket().getRemoteSocketAddress());
                        System.out.println("可读--------------结束");
                        //continue;
                        return;
                    }catch(Exception e){ 
                        System.out.println("error 6,关闭客户端 channel 失败"+e);
                        System.out.println("可读--------------结束");
                        //continue;
                        return;
                    }
    
                }
                byteBuffer.flip();//调用flip之后，读写指针指到缓存头部，并且设置了最多只能读出之前写入的数据长度(而不是整个缓存的容量大小)。
    
                //初始化一个 缓存区可读长度的数组
                byte[] bytes=new byte[byteBuffer.remaining()];
    
                byteBuffer.get(bytes,0,byteBuffer.remaining());
                byteBuffer.clear();
    
                System.out.println("客户端传入信息:"+new String(bytes,CHARSET));
    
                selectionKey.interestOps(SelectionKey.OP_WRITE);//将该 selectionKey 感兴趣事件修改为 写事件，准备向客户端返回数据
                System.out.println("【(selectionKey.isReadable()&&selectionKey.isValid())=true ,已经读取完毕客户端channel 中的buffer信息，将 客户端 channel 感兴趣事件修改为 OP_WRITE】");
                System.out.println("可读--------------结束");
                System.out.println();
            }catch(IOException e){ 
                System.out.println("error 7,出错了");
            }
        }
    
        /** * 处理可写的channel * 向客户端发送信息，该信息从控制台输入 * @param selectionKey */
        private static void writable(SelectionKey selectionKey){ 
            try{ 
                System.out.println("可写--------------开始");
    
                SocketChannel clientChannel=(SocketChannel) selectionKey.channel();//获取对应的客户端 channel
    
                Scanner scanner=new Scanner(System.in);//从控制台输入返回给客户端的信息
                String response=scanner.nextLine()+System.getProperty("line.separator");
    
                int len=clientChannel.write(ByteBuffer.wrap(response.getBytes(CHARSET)));
                if(len==-1){ 
                    try{ 
                        clientChannel.close();
                        selector.wakeup();
                        System.out.println("line 182,客户端已关闭:"+clientChannel.socket().getRemoteSocketAddress());
                        System.out.println("可写--------------结束");
                        //continue;
                        return ;
                    }catch(Exception e){ 
                        System.out.println("error 8"+e);
                        System.out.println("可写--------------结束");
                        //continue;
                        return ;
                    }
                }
    
                selectionKey.interestOps(SelectionKey.OP_READ);//将感兴趣事件修改为 可读
                selector.wakeup();//唤醒其他key
    
                System.out.println("[(selectionKey.isWritable()&&selectionKey.isValid())=true ,服务端返回客户端信息结束, 将客户端的 channel 感兴趣事件修改为 OP_READ】");
                System.out.println("可写--------------结束");
                System.out.println();
            }catch(UnsupportedEncodingException e){ 
                System.out.println("error 9,字符格式不被支持"+e);
            }catch(IOException e2){ 
                System.out.println("error 10"+e2);
            }
        }
    }

###### NIO socket 客户端 ######

> NIO socket 客户端可以请求非 NIO socket 服务端

import java.io.IOException;
    import java.io.UnsupportedEncodingException;
    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.nio.channels.spi.SelectorProvider;
    import java.util.*;
    
    /** * @Title:Socket Client * @Description: NIO Socket * @Author: WuJie */
    public class NioSocketClient { 
        private final static String CHARSET="UTF8";//设置编码格式
        private static Selector selector=null;//nio Selector
    
        public static void main(String [] args){ 
            NioSocketClient nioSocketClient=new NioSocketClient();
            nioSocketClient.connectedServer();//启动客户端，并向服务端发起连接请求
            if(selector!=null){ 
                nioSocketClient.handle(selector);//处理和服务器端的交互
            }else{ 
                System.out.println("error 2,服务端启动失败");
            }
        }
    
        /** * 处理客户端与服务器端的交互 */
        public void connectedServer(){ 
            try{ 
                selector= SelectorProvider.provider().openSelector();//初始化 nio Selector
                selector.select(1000);//阻塞 selector 1000 毫秒，非必须
    
                //Channel 多线程安全、支持异步关闭
                SocketChannel clientChannel=SocketChannel.open();//初始化 客户端 SocketChannel
                clientChannel.configureBlocking(false);//非阻塞模式，否则就不是NIO了
                clientChannel.connect(new InetSocketAddress("127.0.0.1",10000));//连接服务端
    
                clientChannel.register(selector,SelectionKey.OP_CONNECT);//将 clientChannel 注册到selector,感兴趣事件为连接
    
                System.out.println("客户端已启动,客户端字符编码格式为"+CHARSET+",等待取得和服务器端的连接：");
            }catch(IOException e){ 
                System.out.println("error 1,获取 Selector 失败"+e);
            }
        }
    
        /** * 处理和服务器端的交互 * @param selector */
        public void handle(Selector selector){ 
            try{ 
                for(;;){ //客户端始终处于启动状态
                    int keyNumbers=selector.select();//获取已注册 channel 数量
                    if(keyNumbers>0){ //如果已注册 channel 数量>0,则开始处理
                        Set<SelectionKey> selectionKeySet=selector.selectedKeys();//获取 selector 中已注册 channel 的 SelectionKey 集合
    
                        Iterator iterable=selectionKeySet.iterator();//准备遍历 selectionKeySet，处理其对应的 channel
                        while(iterable.hasNext()){ 
                            SelectionKey selectionKey=(SelectionKey)iterable.next();
                            iterable.remove();//Removes from the underlying collection the last element returned by this iterator (optional operation)
    
                            if(selectionKey.isConnectable()){ //可以连接上服务器端
                                connectable(selectionKey);//TODO 可以使用多线程
                            }else if(selectionKey.isReadable()&&selectionKey.isValid()){ //读取服务端传输的数据
                                readable(selectionKey);//TODO 可以使用多线程
                            }else if(selectionKey.isWritable()&&selectionKey.isValid()) { //返回给服务端信息
                                writable(selectionKey);//TODO 可以使用多线程
                            }
    
                        }
                    }else{ 
                        continue;
                    }
                }
    
            }catch(IOException e){ 
                System.out.println("error 3,获取 Selector 已注册 Channel 数量失败"+e);
            }
        }
    
        /** * 处理可连接的channel * 取得和服务端的连接，向服务端发送信息，该信息从控制台输入 * @param selectionKey */
        private static void connectable(SelectionKey selectionKey){ 
            try{ 
                System.out.println("可连接服务端--------------开始");
    
                SocketChannel clientChannel=(SocketChannel)selectionKey.channel();//获取对应的客户端 channel
                if(clientChannel.isConnectionPending()){ //如果正在连接，则完成连接
                    clientChannel.finishConnect();//当 channel 的 socket 连接上后，返回true
                }
                //开始写操作
                Scanner scanner=new Scanner(System.in);//从控制台输入发送给服务端的信息
                String response=scanner.nextLine()+System.getProperty("line.separator");
    
                int len=clientChannel.write(ByteBuffer.wrap(response.getBytes(CHARSET)));
                if(len==-1){ 
                    try{ 
                        clientChannel.close();
                        selector.wakeup();
                        System.out.println("line 122,服务端已关闭:"+clientChannel.socket().getRemoteSocketAddress());
                        System.out.println("可连接服务端--------------结束");
                        //continue;
                        return ;
                    }catch(Exception e){ 
                        System.out.println("error 4"+e);
                        System.out.println("可连接服务端--------------结束");
                        //continue;
                        return ;
                    }
                }
    
                selectionKey.interestOps(SelectionKey.OP_READ);//将感兴趣事件修改为 可读
                selector.wakeup();//唤醒其他key
    
                System.out.println("[(selectionKey.isConnectable()=true ,客户端连接服务端并发送信息结束, 将客户端的 channel 感兴趣事件修改为 OP_READ】");
                System.out.println("可连接服务端--------------结束");
                System.out.println();
            }catch(UnsupportedEncodingException e){ 
                System.out.println("error 5,字符格式不被支持"+e);
            }catch(IOException e2){ 
                System.out.println("error 6"+e2);
            }
    
        }
    
        /** * 处理可读的channel * 读取客户端发送的信息 * @param selectionKey */
        private static void readable(SelectionKey selectionKey){ 
            try{ 
                System.out.println("可读--------------开始");
    
                SocketChannel clientChannel=(SocketChannel) selectionKey.channel();//根据key获取channel
                ByteBuffer byteBuffer= ByteBuffer.allocate(1024); //准备 1k 的缓冲区
                int len=clientChannel.read(byteBuffer);//将channel 内容写入缓存 byteBuffer
                //System.out.println("byteBuffer.remaining()="+byteBuffer.remaining()+";len="+len);
    
                if(len==-1){ //如果没有信息传入，说明客户端可能中断连接，则关闭这个客户端channel
                    try{ 
                        clientChannel.close();
                        selector.wakeup();//唤起其他channel
                        System.out.println("line 160,服务端已关闭:"+clientChannel.socket().getRemoteSocketAddress());
                        System.out.println("可读--------------结束");
                        //continue;
                        return;
                    }catch(Exception e){ 
                        System.out.println("error 7,关闭客户端 channel 失败"+e);
                        System.out.println("可读--------------结束");
                        //continue;
                        return;
                    }
    
                }
                byteBuffer.flip();//调用flip之后，读写指针指到缓存头部，并且设置了最多只能读出之前写入的数据长度(而不是整个缓存的容量大小)。
    
                //初始化一个 缓存区可读长度的数组
                byte[] bytes=new byte[byteBuffer.remaining()];
    
                byteBuffer.get(bytes,0,byteBuffer.remaining());
                byteBuffer.clear();
    
                System.out.println("服务端传入信息:"+new String(bytes,CHARSET));
    
                selectionKey.interestOps(SelectionKey.OP_WRITE);//将该 selectionKey 感兴趣事件修改为 写事件，准备向客户端返回数据
                System.out.println("【(selectionKey.isReadable()&&selectionKey.isValid())=true ,已经读取完毕服务端channel 中的buffer信息，将 客户端 channel 感兴趣事件修改为 OP_WRITE】");
                System.out.println("可读--------------结束");
                System.out.println();
            }catch(IOException e){ 
                System.out.println("error 8,出错了");
            }
        }
    
        /** * 处理可写的channel * 向服务端发送信息，该信息从控制台输入 * @param selectionKey */
        private static void writable(SelectionKey selectionKey){ 
            try{ 
                System.out.println("可写--------------开始");
    
                SocketChannel clientChannel=(SocketChannel) selectionKey.channel();//获取对应的客户端 channel
    
                Scanner scanner=new Scanner(System.in);//从控制台输入返回给客户端的信息
                String response=scanner.nextLine()+System.getProperty("line.separator");
    
                int len=clientChannel.write(ByteBuffer.wrap(response.getBytes(CHARSET)));
                if(len==-1){ 
                    try{ 
                        clientChannel.close();
                        selector.wakeup();
                        System.out.println("line 210,服务端已关闭:"+clientChannel.socket().getRemoteSocketAddress());
                        System.out.println("可写--------------结束");
                        //continue;
                        return ;
                    }catch(Exception e){ 
                        System.out.println("error 9"+e);
                        System.out.println("可写--------------结束");
                        //continue;
                        return ;
                    }
                }
    
                selectionKey.interestOps(SelectionKey.OP_READ);//将感兴趣事件修改为 可读
                selector.wakeup();//唤醒其他key
    
                System.out.println("[(selectionKey.isWritable()&&selectionKey.isValid())=true ,可会淡返回服务端信息结束, 将客户端的 channel 感兴趣事件修改为 OP_READ】");
                System.out.println("可写--------------结束");
                System.out.println();
            }catch(UnsupportedEncodingException e){ 
                System.out.println("error 9,字符格式不被支持"+e);
            }catch(IOException e2){ 
                System.out.println("error 10"+e2);
            }
        }
    }

##### 控制台信息 #####

###### 服务端 ######

服务端已启动,服务器端字符编码格式为UTF8,等待客户端的访问：
    可接收--------------开始
    【selectionKey.isAcceptable()=true，从select 获取 服务端channel ，从这个channel 获取客户端 channel，向 selector 注册 客户端 channel,感兴趣事件为 OP_READ】
    可接收--------------结束
    
    可读--------------开始
    客户端传入信息:第一次请求
    
    【(selectionKey.isReadable()&&selectionKey.isValid())=true ,已经读取完毕客户端channel 中的buffer信息，将 客户端 channel 感兴趣事件修改为 OP_WRITE】
    可读--------------结束
    
    可写--------------开始
    请一次回复
    [(selectionKey.isWritable()&&selectionKey.isValid())=true ,服务端返回客户端信息结束, 将客户端的 channel 感兴趣事件修改为 OP_READ】
    可写--------------结束
    
    可读--------------开始
    客户端传入信息:第二次请求
    
    【(selectionKey.isReadable()&&selectionKey.isValid())=true ,已经读取完毕客户端channel 中的buffer信息，将 客户端 channel 感兴趣事件修改为 OP_WRITE】
    可读--------------结束
    
    可写--------------开始
    第二次回复
    [(selectionKey.isWritable()&&selectionKey.isValid())=true ,服务端返回客户端信息结束, 将客户端的 channel 感兴趣事件修改为 OP_READ】
    可写--------------结束

###### 客户端 ######

客户端已启动,客户端字符编码格式为UTF8,等待取得和服务器端的连接：
    可连接服务端--------------开始
    第一次请求
    [(selectionKey.isConnectable()=true ,客户端连接服务端并发送信息结束, 将客户端的 channel 感兴趣事件修改为 OP_READ】
    可连接服务端--------------结束
    
    可读--------------开始
    服务端传入信息:请一次回复
    
    【(selectionKey.isReadable()&&selectionKey.isValid())=true ,已经读取完毕服务端channel 中的buffer信息，将 客户端 channel 感兴趣事件修改为 OP_WRITE】
    可读--------------结束
    
    可写--------------开始
    第二次请求
    [(selectionKey.isWritable()&&selectionKey.isValid())=true ,可会淡返回服务端信息结束, 将客户端的 channel 感兴趣事件修改为 OP_READ】
    可写--------------结束
    
    可读--------------开始
    服务端传入信息:第二次回复
    
    【(selectionKey.isReadable()&&selectionKey.isValid())=true ,已经读取完毕服务端channel 中的buffer信息，将 客户端 channel 感兴趣事件修改为 OP_WRITE】
    可读--------------结束

[socket_NIO socket]: https://blog.csdn.net/bestcxx/article/details/83447699