• Java 网络编程之TCP(五)：分析服务端注册OP_WRITE写数据的各种场景（三）

    在服务端使用多线程对同个客户端进行读写，会带来意想不到的问题。

    前面的文章中，服务端都是在一个单线程main中，处理所有接收到的IO事件，为了提高效率，会自然的想到，为OP_READ和OP_WRITE事件分配多线程处理。

    需求：服务端把接收到的数据，原样返回给客户端

    服务端代码如下：

    直接在单线程的代码上，把单线程的read和write逻辑，放入一个单独的线程

    服务代码如下：

    1. import java.io.IOException;
    2. import java.net.InetSocketAddress;
    3. import java.nio.ByteBuffer;
    4. import java.nio.channels.SelectionKey;
    5. import java.nio.channels.Selector;
    6. import java.nio.channels.ServerSocketChannel;
    7. import java.nio.channels.SocketChannel;
    8. import java.util.Iterator;
    9. import java.util.Set;
    10.  
    11. public class SocketMultiplexingSingleThreadv2_2 {
    12.  
    13.     private ServerSocketChannel server = null;
    14.     private Selector selector = null;   //linux 多路复用器（select poll epoll） nginx  event{}
    15.     int port = 9090;
    16.  
    17.     public void initServer() {
    18.         try {
    19.             server = ServerSocketChannel.open();
    20.             server.configureBlocking(false);
    21.             server.bind(new InetSocketAddress(port));
    22.             selector = Selector.open();  //  select  poll  *epoll
    23.             server.register(selector, SelectionKey.OP_ACCEPT);
    24.         } catch (IOException e) {
    25.             e.printStackTrace();
    26.         }
    27.     }
    28.  
    29.     public void start() {
    30.         initServer();
    31.         System.out.println("服务器启动了。。。。。");
    32.         try {
    33.             while (true) {
    34. //                Set keys = selector.keys();
    35. //                System.out.println(keys.size()+"   size");
    36.                 while (selector.select(50) > 0) {
    37.                     Set selectionKeys = selector.selectedKeys();
    38.                     Iterator iter = selectionKeys.iterator();
    39.                     while (iter.hasNext()) {
    40.                         SelectionKey key = iter.next();
    41.                         iter.remove();
    42.                         if (key.isAcceptable()) {
    43.                             acceptHandler(key);
    44.                         } else if (key.isReadable()) {
    45. //                            key.cancel();  //现在多路复用器里把key  cancel了
    46.                             System.out.println("in.....");
    47.                             readHandler(key);//还是阻塞的嘛？ 即便以抛出了线程去读取，但是在时差里，这个key的read事件会被重复触发
    48.  
    49.                         } else if(key.isWritable()){  //我之前没讲过写的事件！！！！！
    50.                             //写事件<--  send-queue  只要是空的，就一定会给你返回可以写的事件，就会回调我们的写方法
    51.                             //你真的要明白：什么时候写？不是依赖send-queue是不是有空间
    52.                             //1，你准备好要写什么了，这是第一步
    53.                             //2，第二步你才关心send-queue是否有空间
    54.                             //3，so，读 read 一开始就要注册，但是write依赖以上关系，什么时候用什么时候注册
    55.                             //4，如果一开始就注册了write的事件，进入死循环，一直调起！！！
    56. //                            key.cancel();
    57.                             key.interestOps(key.interestOps() & ~SelectionKey.OP_WRITE);
    58.  
    59.  
    60.  
    61.                             writeHandler(key);
    62.                         }
    63.                     }
    64.                 }
    65.             }
    66.         } catch (IOException e) {
    67.             e.printStackTrace();
    68.         }
    69.     }
    70.  
    71.     private void writeHandler(SelectionKey key) {
    72.         new Thread(()->{
    73.             System.out.println("write handler...");
    74.             SocketChannel client = (SocketChannel) key.channel();
    75.             ByteBuffer buffer = (ByteBuffer) key.attachment();
    76.             buffer.flip();
    77.             while (buffer.hasRemaining()) {
    78.                 try {
    79.                     int write = client.write(buffer);
    80.                     System.out.println("write " + Thread.currentThread().getName()+ " " + write);
    81.                 } catch (IOException e) {
    82.                     e.printStackTrace();
    83.                 }
    84.             }
    85.             try {
    86.                 Thread.sleep(2000);
    87.             } catch (InterruptedException e) {
    88.                 e.printStackTrace();
    89.             }
    90.             buffer.clear();
    91. //            key.cancel();
    92.  
    93. //            try {
    94.                 client.shutdownOutput();
    95. //
    96.                 client.close();
    97. //
    98. //            } catch (IOException e) {
    99. //                e.printStackTrace();
    100. //            }
    101.         }).start();
    102.  
    103.     }
    104.  
    105.     public void acceptHandler(SelectionKey key) {
    106.         try {
    107.             ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
    108.             SocketChannel client = ssc.accept();
    109.             client.configureBlocking(false);
    110.             ByteBuffer buffer = ByteBuffer.allocate(8192);
    111.             client.register(selector, SelectionKey.OP_READ, buffer);
    112.             System.out.println("-------------------------------------------");
    113.             System.out.println("新客户端：" + client.getRemoteAddress());
    114.             System.out.println("-------------------------------------------");
    115.  
    116.         } catch (IOException e) {
    117.             e.printStackTrace();
    118.         }
    119.     }
    120.  
    121.     public void readHandler(SelectionKey key) {
    122.         new Thread(()->{
    123.             System.out.println("read handler.....");
    124.             SocketChannel client = (SocketChannel) key.channel();
    125.             ByteBuffer buffer = (ByteBuffer) key.attachment();
    126.             buffer.clear();
    127.             int read = 0;
    128.             try {
    129.                 while (true) {
    130.                     read = client.read(buffer);
    131.                     System.out.println("read " + Thread.currentThread().getName()+ " " + read);
    132.                     if (read > 0) {
    133.                         client.register(key.selector(), key.interestOps() + SelectionKey.OP_WRITE,buffer);
    134.                     } else if (read == 0) {
    135.  
    136.                         break;
    137.                     } else {
    138.                         client.close();
    139.                         break;
    140.                     }
    141.                 }
    142.             } catch (IOException e) {
    143.                 try {
    144.                     System.out.println("client " + client.getRemoteAddress() + " disconnected");
    145.                     client.close();
    146.                 } catch (IOException ex) {
    147.                     throw new RuntimeException(ex);
    148.                 }
    149.                 e.printStackTrace();
    150.             }
    151.         }).start();
    152.  
    153.     }
    154.  
    155.     public static void main(String[] args) {
    156.         SocketMultiplexingSingleThreadv2_2 service = new SocketMultiplexingSingleThreadv2_2();
    157.         service.start();
    158.     }
    159. }

    测试：

    先启动一个服务端，再启动一个客户端，客户端发送数据

    服务端日志：

    1. 服务器启动了。。。。。
    2. -------------------------------------------
    3. 新客户端：/127.0.0.1:21598
    4. -------------------------------------------
    5. in.....
    6. in.....
    7. read handler.....
    8. in.....
    9. read handler.....
    10. read Thread-0 5
    11. read Thread-1 0
    12. read handler.....
    13. read Thread-2 0
    14. read Thread-0 0
    15. write handler...

    客户端日志：

    1. client connected to server
    2. 1234
    3. client receive data from consolejava.io.BufferedInputStream@6acfcaf3 : 1234
    4.

    可以看到，客户端发送数据，没有接收到服务端返回的数据；

    服务端接收到数据后，在写数据的时候，buffer中没有数据可写；

    再仔细看下服务端的日志，可以同个客户端只发送一条数据的时候，有3个线程来处理，其他两个线程读到的数据都是0；


    一个客户端的读事件，分配一个线程处理，但是线程还没处理完，下个读事件就来了，就又分配一个线程处理。。。而同一个客户端共享一个buffer，在register OP_READ的时候attach的。
    这样使得buffer中的数据还没来得及写出去，就被其他读线程给冲掉了（read == 0）；

    tip:read事件来的时候，如果不读取数据，read事件会一直有的

    解决方法：不可以并发读同一个client， 在处理一个Client的 OP_READ的时候先取消 OP_READ的注册，读完了后，在注册一个 OP_READ

    新的服务端代码：

    1. import java.io.IOException;
    2. import java.net.InetSocketAddress;
    3. import java.nio.ByteBuffer;
    4. import java.nio.channels.*;
    5. import java.util.Iterator;
    6. import java.util.Set;
    7.  
    8. public class SocketMultiplexingSingleThreadv2 {
    9.  
    10.     private ServerSocketChannel server = null;
    11.     private Selector selector = null;   //linux 多路复用器（select poll epoll） nginx  event{}
    12.     int port = 9090;
    13.  
    14.     public void initServer() {
    15.         try {
    16.             server = ServerSocketChannel.open();
    17.             server.configureBlocking(false);
    18.             server.bind(new InetSocketAddress(port));
    19.             selector = Selector.open();  //  select  poll  *epoll
    20.             server.register(selector, SelectionKey.OP_ACCEPT);
    21.         } catch (IOException e) {
    22.             e.printStackTrace();
    23.         }
    24.     }
    25.  
    26.     public void start() {
    27.         initServer();
    28.         System.out.println("服务器启动了。。。。。");
    29.         try {
    30.             while (true) {
    31. //                Set keys = selector.keys();
    32. //                System.out.println(keys.size()+"   size");
    33.                 while (selector.select(50) > 0) {
    34.                     Set selectionKeys = selector.selectedKeys();
    35.                     Iterator iter = selectionKeys.iterator();
    36.                     while (iter.hasNext()) {
    37.                         SelectionKey key = iter.next();
    38.                         iter.remove();
    39.                         if (key.isAcceptable()) {
    40.                             acceptHandler(key);
    41.                         } else if (key.isReadable()) {
    42. //                            key.cancel();  //现在多路复用器里把key  cancel了
    43.                             System.out.println("in.....");
    44.                             // 同一个Client，读之前先取消OP_READ，防止多线程冲突吹
    45.                             key.interestOps(key.interestOps() & ~SelectionKey.OP_READ);
    46.  
    47.                             readHandler(key);//还是阻塞的嘛？ 即便以抛出了线程去读取，但是在时差里，这个key的read事件会被重复触发
    48.  
    49.                         } else if(key.isWritable()){  //我之前没讲过写的事件！！！！！
    50.                             //写事件<--  send-queue  只要是空的，就一定会给你返回可以写的事件，就会回调我们的写方法
    51.                             //你真的要明白：什么时候写？不是依赖send-queue是不是有空间
    52.                             //1，你准备好要写什么了，这是第一步
    53.                             //2，第二步你才关心send-queue是否有空间
    54.                             //3，so，读 read 一开始就要注册，但是write依赖以上关系，什么时候用什么时候注册
    55.                             //4，如果一开始就注册了write的事件，进入死循环，一直调起！！！
    56. //                            key.cancel();
    57.                             key.interestOps(key.interestOps() & ~SelectionKey.OP_WRITE);
    58.  
    59.  
    60.  
    61.                             writeHandler(key);
    62.                         }
    63.                     }
    64.                 }
    65.             }
    66.         } catch (IOException e) {
    67.             e.printStackTrace();
    68.         }
    69.     }
    70.  
    71.     private void writeHandler(SelectionKey key) {
    72.         new Thread(()->{
    73.             System.out.println("write handler...");
    74.             SocketChannel client = (SocketChannel) key.channel();
    75.             ByteBuffer buffer = (ByteBuffer) key.attachment();
    76.             buffer.flip();
    77.             while (buffer.hasRemaining()) {
    78.                 try {
    79.                     int write = client.write(buffer);
    80.                     System.out.println("write " + Thread.currentThread().getName()+ " " + write);
    81.                 } catch (IOException e) {
    82.                     e.printStackTrace();
    83.                 }
    84.             }
    85.             try {
    86.                 Thread.sleep(2000);
    87.             } catch (InterruptedException e) {
    88.                 e.printStackTrace();
    89.             }
    90.             buffer.clear();
    91. //            key.cancel();
    92.  
    93. //            try {
    94.                 client.shutdownOutput();
    95. //
    96.                 client.close();
    97. //
    98. //            } catch (IOException e) {
    99. //                e.printStackTrace();
    100. //            }
    101.         }).start();
    102.  
    103.     }
    104.  
    105.     public void acceptHandler(SelectionKey key) {
    106.         try {
    107.             ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
    108.             SocketChannel client = ssc.accept();
    109.             client.configureBlocking(false);
    110.             ByteBuffer buffer = ByteBuffer.allocate(8192);
    111.             client.register(selector, SelectionKey.OP_READ, buffer);
    112.             System.out.println("-------------------------------------------");
    113.             System.out.println("新客户端：" + client.getRemoteAddress());
    114.             System.out.println("-------------------------------------------");
    115.  
    116.         } catch (IOException e) {
    117.             e.printStackTrace();
    118.         }
    119.     }
    120.  
    121.     public void readHandler(SelectionKey key) {
    122.         new Thread(()->{
    123.             System.out.println("read handler.....");
    124.             SocketChannel client = (SocketChannel) key.channel();
    125.             ByteBuffer buffer = (ByteBuffer) key.attachment();
    126.             buffer.clear();
    127.             int read = 0;
    128.             try {
    129.                 while (true) {
    130.                     read = client.read(buffer);
    131.                     System.out.println(Thread.currentThread().getName()+ " " + read);
    132.                     if (read > 0) {
    133.                         // 同一个Client，读完数据后，要再次注册OP_READ，读后面发送过来的数据
    134.                         key.interestOps(  SelectionKey.OP_READ);
    135.  
    136.                         client.register(key.selector(), key.interestOps() + SelectionKey.OP_WRITE,buffer);
    137.                     } else if (read == 0) {
    138.  
    139.                         break;
    140.                     } else {
    141.                         client.close();
    142.                         break;
    143.                     }
    144.                 }
    145.             } catch (IOException e) {
    146.                 try {
    147.                     System.out.println("client " + client.getRemoteAddress() + " disconnected");
    148.                     client.close();
    149.                 } catch (IOException ex) {
    150.                     throw new RuntimeException(ex);
    151.                 }
    152.                 e.printStackTrace();
    153.             }
    154.         }).start();
    155.  
    156.     }
    157.  
    158.     public static void main(String[] args) {
    159.         SocketMultiplexingSingleThreadv2 service = new SocketMultiplexingSingleThreadv2();
    160.         service.start();
    161.     }
    162. }

    测试：

    先启动一个服务端，再启动一个客户端1，客户端1发送数据

    服务端日志：

    1. 服务器启动了。。。。。
    2. -------------------------------------------
    3. 新客户端：/127.0.0.1:24029
    4. -------------------------------------------
    5. in.....
    6. read handler.....
    7. Thread-0 8
    8. Thread-0 0
    9. write handler...
    10. write Thread-1 8

    客户端1日志：

    1. client connected to server
    2. client1
    3. client receive data from consolejava.io.BufferedInputStream@65231a33 : client1
    4.  
    5. client receive data from serverjava.net.Socket$SocketInputStream@4629104a data size：8: client1

    可以看到，客户单和服务端都可以正常接收和发送数据。

    再添加一个客户端2，发送数据

    服务端日志：

    1. 服务器启动了。。。。。
    2. -------------------------------------------
    3. 新客户端：/127.0.0.1:24029
    4. -------------------------------------------
    5. in.....
    6. read handler.....
    7. Thread-0 8
    8. Thread-0 0
    9. write handler...
    10. write Thread-1 8
    11. -------------------------------------------
    12. 新客户端：/127.0.0.1:24105
    13. -------------------------------------------
    14. in.....
    15. read handler.....
    16. Thread-2 8
    17. Thread-2 0
    18. write handler...
    19. write Thread-3 8

    客户端2的日志：

    1. client connected to server
    2. client2
    3. client receive data from consolejava.io.BufferedInputStream@65231a33 : client2
    4.  
    5. client receive data from serverjava.net.Socket$SocketInputStream@27f8302d data size：8: client2

    可以看到，客户端2和服务端都可以正常接收和发送数据。

    客户端1，再次发送数据

    客户端日志：

    1. client connected to server
    2. client1
    3. client receive data from consolejava.io.BufferedInputStream@65231a33 : client1
    4.  
    5. client receive data from serverjava.net.Socket$SocketInputStream@4629104a data size：8: client1
    6.  
    7. clent1_2
    8. client receive data from consolejava.io.BufferedInputStream@65231a33 : clent1_2
    9.  
    10. client receive data from serverjava.net.Socket$SocketInputStream@4629104a data size：9: clent1_2

    服务端日志：

    1. 服务器启动了。。。。。
    2. -------------------------------------------
    3. 新客户端：/127.0.0.1:24029
    4. -------------------------------------------
    5. in.....
    6. read handler.....
    7. Thread-0 8
    8. Thread-0 0
    9. write handler...
    10. write Thread-1 8
    11. -------------------------------------------
    12. 新客户端：/127.0.0.1:24105
    13. -------------------------------------------
    14. in.....
    15. read handler.....
    16. Thread-2 8
    17. Thread-2 0
    18. write handler...
    19. write Thread-3 8
    20. in.....
    21. read handler.....
    22. Thread-4 9
    23. Thread-4 0
    24. write handler...
    25. write Thread-5 9

    从服务端日志中，可以看到，每个客户端的读事件，只有一个线程处理。

    整个处理流程是服务预期的。

  • 相关阅读:
    函数——两个数的合并
    One Last Kiss风格封面生成器；程序内存分析工具；Python入门课程资料；神经文本语音合成教程；前沿论文 | ShowMeAI资讯日报
    Promise笔记-同步回调-异步回调-JS中的异常error处理-Promis的理解和使用-基本使用-链式调用-七个关键问题
    课题学习(九)----阅读《导向钻井工具姿态动态测量的自适应滤波方法》论文笔记
    什么是RESTful API，Spring MVC如何支持RESTful架构
    麒麟系统安装找不到安装源！！！！设置基础软件仓库时出错
    复杂分数 马蹄集
    CTO 说了，用错@Autowired 和@Resource 的人可以领盒饭了
    Java 18 新功能介绍
    使用ElementUI结合Mock完成主页的搭建
  • 原文地址：https://blog.csdn.net/u013771019/article/details/138182198