devbox
小蓝xlanll
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

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

作者：小蓝xlanll | 2024-04-26 10:21:29

赞

踩

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<SelectionKey> keys = selector.keys();
  35. //                System.out.println(keys.size()+"   size");
  36.                 while (selector.select(50) > 0) {
  37.                     Set<SelectionKey> selectionKeys = selector.selectedKeys();
  38.                     Iterator<SelectionKey> 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<SelectionKey> keys = selector.keys();
  32. //                System.out.println(keys.size()+"   size");
  33.                 while (selector.select(50) > 0) {
  34.                     Set<SelectionKey> selectionKeys = selector.selectedKeys();
  35.                     Iterator<SelectionKey> 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

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

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

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/小蓝xlanll/article/detail/490393
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号