java nio非阻塞式网络通信入门案例（nio服务端与bio多线程客户端(java/python）

我会带着你远行 2022-09-11 11:25 22阅读 0赞

###  ###

*   *  nio服务端：
     *  改进服务端
     *  java客户端
     *  python版本客户端：
     *  python客户端改进版（多线程执行）

注意：如果想把服务端程序放在自己的服务器上，要记得开放相应的端口，否则客户端会显示连接超时。socket的ip地址改为公网ip,

## nio服务端： ##

/** * @Author * @Description 学习selector * @Date **/
    public class select { 
        public void getSelector() throws IOException { 
            //创建 Selector
            Selector selector = Selector.open();//用open方法创建
            ServerSocketChannel channel = ServerSocketChannel.open();//获取通道
            channel.configureBlocking(false);//切换为非阻塞模式
            channel.bind(new InetSocketAddress(9999));
            channel.register(selector, SelectionKey.OP_ACCEPT);//注册通道到选择器上，第二个参数为指定的事件为”监听接收事件“
    //
    // * 读 : SelectionKey.OP_READ （1）
    //* 写 : SelectionKey.OP_WRITE （4）
    //* 连接 : SelectionKey.OP_CONNECT （8）
    //* 接收 : SelectionKey.OP_ACCEPT （16）
    //* 若注册时不止监听一个事件，则可以使用“位或”操作符连接。
    
            //轮询式的获取选择器上已经“准备就绪”的事件
            while (selector.select() > 0) { 
                System.out.println("开始");
                //7. 获取当前选择器中所有注册的“选择键(已就绪的监听事件)”
                Iterator<SelectionKey> it = selector.selectedKeys().iterator();
                while (it.hasNext()) { 
                    //8. 获取准备“就绪”的是事件
                    SelectionKey sk = it.next();
                    //9. 判断具体是什么事件准备就绪
                    if (sk.isAcceptable()) { 
                        //10. 若“接收就绪”，获取客户端连接
                        SocketChannel sChannel = channel.accept();
                        //11. 切换非阻塞模式
                        sChannel.configureBlocking(false);
                        //12. 将该通道注册到选择器上
                        sChannel.register(selector, SelectionKey.OP_READ);
                    } else if (sk.isReadable()) { 
                        //13. 获取当前选择器上“读就绪”状态的通道
                        SocketChannel sChannel = (SocketChannel) sk.channel();
                        //14. 读取数据
                        ByteBuffer buf = ByteBuffer.allocate(1024);
                        int len = 0;
                        while ((len = sChannel.read(buf)) > 0) { 
                            buf.flip();
                            System.out.println(new String(buf.array(), 0, len));
                            buf.clear();
                        }
                    }
                    //15. 取消选择键 SelectionKey
                    it.remove();
                }
            }
        }
    
        public static void main(String[] args) throws IOException { 
            new select().getSelector();
        }
    }

## 改进服务端 ##

加一个catch ioexception，防止因为客户端关闭连接导致服务端也异常终止  
主要逻辑在读取数据的函数 readData(sk);中，其他部分跟上面一样

while (selector.select() > 0) { 
              // System.out.println("开始：");
                //7. 获取当前选择器中所有注册的“选择键(已就绪的监听事件)”
                Iterator<SelectionKey> it = selector.selectedKeys().iterator();
                while (it.hasNext()) { 
                    //8. 获取准备“就绪”的是事件
                    SelectionKey sk = it.next();
                    //9. 判断具体是什么事件准备就绪
                    if (sk.isAcceptable()) { 
                     // 10. 若“接收就绪”，获取客户端连接
                        SocketChannel sChannel = channel.accept();
                        //11. 切换非阻塞模式
                        sChannel.configureBlocking(false);
                        //12. 将该通道注册到选择器上
                        sChannel.register(selector, SelectionKey.OP_READ);
                    } else if (sk.isReadable()) { 
                         readData(sk);
                        //15. 取消选择键 SelectionKey
                    }
                        it.remove();
                    }
                }
            }
    
     //读取客户端消息
            private void readData (SelectionKey key){ 
                //取到关联的channle
                SocketChannel channel = null;
                try { 
                    //得到channel
                    channel = (SocketChannel) key.channel();
                    //创建buffer
                    ByteBuffer buffer = ByteBuffer.allocate(1024);
                    int count = channel.read(buffer);
                    //根据count的值做处理
                    if (count > 0) { 
                        //把缓存区的数据转成字符串
                        String msg = new String(buffer.array());
                        //输出该消息
                        System.out.println("form 客户端: " + msg);
                        //向其它的客户端转发消息(去掉自己), 专门写一个方法来处理
                        // sendInfoToOtherClients(msg, channel);
                    }
                    else { 
                        System.out.println("no dara");
                    }
                } catch (IOException e) { 
                    try { 
                        System.out.println(channel.getRemoteAddress() + " 离线了..");
                        e.printStackTrace();
                        //取消注册
                        key.cancel();
                        //关闭通道
                        channel.close();
                    } catch (IOException e2) { 
                        e2.printStackTrace();
                        ;
                    }
                }
            }

这样的话会发现即使服务端发生io异常也仍然未终止程序，有新的客户端连接时还可以继续正常运行  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAbWFyY2ggb2YgVGltZQ_size_20_color_FFFFFF_t_70_g_se_x_16]

## java客户端 ##

客户端模拟多线程发送socket数据：

/** * @Author * @Description 多线程bio读写数据 * @Date **/
    
    public class mutiClient { 
    
        public static void main(String[] args) throws InterruptedException { 
            // 接下来模拟3个Client并发访问服务器
    
            int poolsize = 3;
    
    
    
            ExecutorService pool = Executors.newFixedThreadPool(poolsize);
    
            Collection< Callable<Object>> tasks =new ArrayList<>(10);
            final String clientname="clientThread";
    
            for (int i = 0; i < poolsize; i++) { 
    
                final int n = i;
                // 若每一个Client都保持使用BIO方式发送数据到Server，并读取数据。
    
                tasks.add(new Callable() { 
                    @Override
    
                    public Object call() throws Exception { 
                        Socket socket = new Socket("127.0.0.1", 9999);
                        final InputStream input = socket.getInputStream();
                        final OutputStream out = socket.getOutputStream();
                        final String clientname_n = clientname + "_" + n;
    
                        // BIO读取数据线程
    
                        new Thread(clientname_n + "_read") { 
                            @Override
                            public void run() { 
    
                                byte[] bs = new byte[1024];
    
                                while (true) { 
    
                                    try { 
                                        Thread.sleep(1000);
    
                                    } catch (InterruptedException e) { 
                                        e.printStackTrace();
                                    }
                                    int len = 0;
                                    try { 
                                        while ((len = input.read(bs)) != -1) { 
                                            System.out.println("Clinet thread " + Thread.currentThread()
                                                    .getName() + " read: " + new String(bs, 0, len));
    
                                        }
    
                                    } catch (IOException e) { 
                                        e.printStackTrace();
                                    }
                                }
                            }
                        }.start();
    
    
    
                        // BIO写数据线程
    
                        new Thread(clientname_n + "_write") { 
    
                            @Override
                            public void run() { 
                                int a = 0;
                                while (true) { 
                                    try { 
                                        Thread.sleep(100);
                                    } catch (InterruptedException e) { 
                                        e.printStackTrace();
                                    }
                                    String str = Thread.currentThread().getName()
                                            + " hello, " + a;
                                    try { 
                                        out.write(str.getBytes());
                                        a++;
                                    } catch (IOException e) { 
                                        e.printStackTrace();
                                    }
                                }
                            }
                        }.start();
                        return null;
    
                    }
                });
    
            }
    pool.invokeAll(tasks);

运行服务端再运行客户端，j结果：  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAbWFyY2ggb2YgVGltZQ_size_16_color_FFFFFF_t_70_g_se_x_16]

## python版本客户端： ##

from socket import *
    host='localhost'
    port=9999
    bufsize=1024
    addr=(host,port)
    tcpClient=socket(AF_INET,SOCK_STREAM)
    tcpClient.connect(addr)
    while True:
        data=input('>')
        if not data:
            break
        tcpClient.send(str.encode(data))
        data=tcpClient.recv(bufsize)
        if not data:
            continue
        print(data)
    tcpClient.close()

结果：  
![在这里插入图片描述][2b5ad39129504d6faf934f1981a96fdc.png]  
![在这里插入图片描述][59d7f563fd0f4fb284a9dfca9e8826f6.png]

## python客户端改进版（多线程执行） ##

注意：写sleep是因为防止主线程结束了导致多线程没有执行而失效

def tcpcilent(threadName,port):
        host = 'localhost'
        bufsize = 1024
        port=9999
        addr = (host, port)
        tcpClient = socket(AF_INET, SOCK_STREAM)
        tcpClient.connect(addr)
        i  =0
        #tcpClient.setblocking(0)
        while True:
            tcpClient.send(str.encode(threadName+"uiyuiyiu"+'\n'))
          
            time.sleep(1)
            if i>100:
                break
        tcpClient.close()
    if __name__ == '__main__':
    
       try:
            _thread.start_new_thread(tcpcilent, ("thread1", 9999,))
            _thread.start_new_thread(tcpcilent, ("thread2", 9999,))
       except:
            print("无法启动线程")
       time.sleep(5)

服务端：  
![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAbWFyY2ggb2YgVGltZQ_size_18_color_FFFFFF_t_70_g_se_x_16]

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAbWFyY2ggb2YgVGltZQ_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20220828/e8cbabd322c74fbeaf66463617bf0568.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAbWFyY2ggb2YgVGltZQ_size_16_color_FFFFFF_t_70_g_se_x_16]: /images/20220828/28284d9c371d4b9380c193cfb7d08629.png
[2b5ad39129504d6faf934f1981a96fdc.png]: /images/20220828/da2a5989e30d4dde96c59b0a9eebfb80.png
[59d7f563fd0f4fb284a9dfca9e8826f6.png]: /images/20220828/704df6acf3ab4ff1b933be8f8917af5c.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBAbWFyY2ggb2YgVGltZQ_size_18_color_FFFFFF_t_70_g_se_x_16]: /images/20220828/2e4cd4d0c8f5494882898db143277211.png