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

    在前面的文章中，我们分析了在NIO模式下，服务端接收和读取多个客户端数据的情况；

    本文，我们看下NIO模式下，使用Selector如何把数据发送给客户端。

    一、直接通过java.nio.channels.SocketChannel#write(java.nio.ByteBuffer)方法写

    二、注册OP_WRITE，在写状态满足情况下，通过java.nio.channels.SocketChannel#write(java.nio.ByteBuffer)方法写

    我们看下第一种情况：

    服务端在有数据需要发送给客户端的情况下，直接使用客户端对应的SocketChannel写数据即可，不考虑数据是否可以正常发送。

    这和BIO下，直接通过输出流写数据是一样的

    我们可以基于聊天室的功能，把收到的数据直接写给其他客户端

    客户端代码统一使用BIO，如下：

    1. import java.io.IOException;
    2. import java.io.InputStream;
    3. import java.io.OutputStream;
    4. import java.net.Socket;
    5.  
    6. /**
    7.  * 基于BIO的TCP网络通信的客户端，接收控制台输入的数据，然后通过字节流发送给服务端
    8.  *
    9.  */
    10. class ChatClient {
    11.     public static void main(String[] args) throws IOException {
    12.         // 连接server
    13.         Socket serverSocket = new Socket("localhost", 9090);
    14.         System.out.println("client connected to server");
    15.  
    16.         // 读取用户在控制台上的输入，并发送给服务器
    17.         new Thread(new ClientThread(serverSocket)).start();
    18.  
    19.         // 接收服务端发送过来的数据
    20.         try (InputStream serverSocketInputStream = serverSocket.getInputStream();) {
    21.             byte[] buffer = new byte[1024];
    22.             int len;
    23.             while ((len = serverSocketInputStream.read(buffer)) != -1) {
    24.                 String data = new String(buffer, 0, len);
    25.                 System.out.println(
    26.                     "client receive data from server" + serverSocketInputStream + " data size：" + len + ": " + data);
    27.             }
    28.         }
    29.  
    30.     }
    31. }
    32.  
    33. class ClientThread implements Runnable {
    34.     private Socket serverSocket;
    35.  
    36.     public ClientThread(Socket serverSocket) {
    37.         this.serverSocket = serverSocket;
    38.     }
    39.  
    40.     @Override
    41.     public void run() {
    42.         // 读取用户在控制台上的输入，并发送给服务器
    43.         InputStream in = System.in;
    44.         byte[] buffer = new byte[1024];
    45.         int len;
    46.         try (OutputStream outputStream = serverSocket.getOutputStream();) {
    47.             // read操作阻塞，直到有数据可读，由于后面还要接收服务端转发过来的数据，这两个操作都是阻塞的，所以需要两个线程
    48.             while ((len = in.read(buffer)) != -1) {
    49.                 String data = new String(buffer, 0, len);
    50.                 System.out.println("client receive data from console" + in + " : " + new String(buffer, 0, len));
    51.                 if ("exit\n".equals(data)) {
    52.                     // 模拟客户端关闭连接
    53.                     System.out.println("client close :" + serverSocket);
    54.                     // 这里跳出循环后，try-with-resources 会自动关闭outputStream
    55.                     break;
    56.                 }
    57.                 // 发送数据给服务器端
    58.                 outputStream.write(new String(buffer, 0, len).getBytes()); // 此时buffer中是有换行符
    59.             }
    60.         } catch (IOException e) {
    61.             throw new RuntimeException(e);
    62.         }
    63.     }
    64. }

    服务端代码如下：

    1. import java.io.IOException;
    2. import java.net.InetSocketAddress;
    3. import java.nio.ByteBuffer;
    4. import java.nio.channels.SelectableChannel;
    5. import java.nio.channels.SelectionKey;
    6. import java.nio.channels.Selector;
    7. import java.nio.channels.ServerSocketChannel;
    8. import java.nio.channels.SocketChannel;
    9. import java.util.Iterator;
    10. import java.util.Set;
    11.  
    12. /**
    13.  * 基于NIO实现服务端，通过Selector基于事件驱动客户端的读取
    14.  * 服务端接收到数据后，直接写回客户端，或者写给其他客户端
    15.  *
    16.  */
    17. class NIOSelectorDirectWriteServer {
    18.     Selector selector;
    19.  
    20.     public static void main(String[] args) throws IOException {
    21.         NIOSelectorDirectWriteServer server = new NIOSelectorDirectWriteServer();
    22.         server.start(); // 开启监听和事件处理
    23.     }
    24.  
    25.     public void start() {
    26.         initServer();
    27.         // selector非阻塞轮询有哪些感兴趣的事件到了
    28.         doService();
    29.     }
    30.  
    31.     private void doService() {
    32.         if (selector == null) {
    33.             System.out.println("server init failed, without doing read/write");
    34.             return;
    35.         }
    36.         try {
    37.             while (true) {
    38.                 while (selector.select() > 0) {
    39.                     Set keys = selector.selectedKeys(); // 感兴趣且准备好的事件
    40.                     Iterator iterator = keys.iterator(); // 迭代器遍历处理，后面要删除集合元素
    41.                     while (iterator.hasNext()) {
    42.                         SelectionKey key = iterator.next();
    43.                         iterator.remove(); // 删除当前元素，防止重复处理
    44.                         // 下面根据事件进行分别处理
    45.                         if (key.isAcceptable()) {
    46.                             // 客户端连接事件
    47.                             acceptHandler(key);
    48.                         } else if (key.isReadable()) {
    49.                             // 读取客户端数据
    50.                             readHandler(key);
    51.                         }
    52.                     }
    53.                 }
    54.             }
    55.         } catch (IOException exception) {
    56.             exception.printStackTrace();
    57.         }
    58.     }
    59.  
    60.     private void initServer() {
    61.         try {
    62.             ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
    63.             serverSocketChannel.configureBlocking(false);
    64.             serverSocketChannel.bind(new InetSocketAddress(9090));
    65.  
    66.             // 此时在selector上注册感兴趣的事件
    67.             // 这里先注册OP_ACCEPT: 客户端连接事件
    68.             selector = Selector.open();
    69.             serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
    70.             System.out.println("server init success");
    71.         } catch (IOException exception) {
    72.             exception.printStackTrace();
    73.             System.out.println("server init failied");
    74.         }
    75.     }
    76.  
    77.     public void acceptHandler(SelectionKey key) {
    78.         ServerSocketChannel server = (ServerSocketChannel) key.channel(); // 获取客户端的channel
    79.         try {
    80.             SocketChannel client = server.accept();
    81.             client.configureBlocking(false); // 设置client非阻塞
    82.             System.out.println("server receive a client :" + client);
    83.             // 注册OP_READ事件，用于从客户端读取数据
    84.             // 给Client分配一个buffer，用于读取数据，注意buffer的线程安全
    85.             ByteBuffer buffer = ByteBuffer.allocate(1024); // buffer这个参数注册的时候也可以不用
    86.             client.register(key.selector(), SelectionKey.OP_READ, buffer);
    87.         } catch (IOException exception) {
    88.             exception.printStackTrace();
    89.         }
    90.     }
    91.  
    92.     public void readHandler(SelectionKey key) {
    93.         System.out.println("read handler");
    94.         SocketChannel client = (SocketChannel) key.channel(); // 获取客户端的channel
    95.         ByteBuffer buffer = (ByteBuffer) key.attachment(); // 获取Client channel关联的buffer
    96.         buffer.clear(); // 使用前clear
    97.  
    98.         // 防止数据分包，需要while循环读取
    99.         try {
    100.             while (true) {
    101.                 int readLen = client.read(buffer);
    102.                 if (readLen > 0) {
    103.                     // 读取到数据了
    104.                     buffer.flip();
    105.                     byte[] data = new byte[buffer.limit()];
    106.                     buffer.get(data);
    107.                     System.out.println("server read data from " + client + ", data is :" + new String(data));
    108.                     // 转发给其他客户端
    109.                     forwardMsg(client, data);
    110.                 } else if (readLen == 0) {
    111.                     // 没读到数据
    112.                     System.out.println(client + " : no data");
    113.                     break;
    114.                 } else if (readLen < 0) {
    115.                     // client 关闭连接
    116.                     // 当客户端主动连接断开时，为了让服务器知道断开了连接，会产生OP_READ事件。所以需要判断读取长度，当读到-1时，关闭channel。
    117.                     System.out.println(client + " close");
    118.                     client.close();
    119.                     break;
    120.                 }
    121.             }
    122.         } catch (IOException exception) {
    123.             exception.printStackTrace();
    124.             // client 关闭连接
    125.             System.out.println(client + " disconnect");
    126.             // todo:disconnect 导致一直有read事件，怎么办？
    127.             try {
    128.                 client.close();
    129.             } catch (IOException ex) {
    130.                 System.out.println("close ex");
    131.             }
    132.         }
    133.     }
    134.  
    135.     private void forwardMsg(SocketChannel myself, byte[] msg) throws IOException {
    136.         Set keys = selector.keys();
    137.         // read/write 对应同一个key，同一个client不会发送两遍
    138.         for (SelectionKey key : keys) {
    139.             SelectableChannel channel = key.channel();
    140.             if (channel instanceof SocketChannel && channel != myself) {
    141.                 // 发送数据
    142.                 SocketChannel client = (SocketChannel) channel;
    143.                 System.out.println("forward msg to " + client);
    144.                 ByteBuffer buff = ByteBuffer.wrap(msg);
    145.                 while (buff.hasRemaining()) {
    146.                     client.write(buff);
    147.                 }
    148.             }
    149.         }
    150.     }
    151. }

    可以看到，这里我们在forwardMsg()方法中，直接通过channel 把数据转发出去了

    测试：

    先启动服务器，然后启动两个客户端

    客户端1发送消息client1，服务端接收到后转发给客户端2

    客户端2发送消息client2，服务端接收到后转发给客户端1

    服务端日志：

    1. server init success
    2. server receive a client :java.nio.channels.SocketChannel[connected local=/127.0.0.1:9090 remote=/127.0.0.1:23536]
    3. server receive a client :java.nio.channels.SocketChannel[connected local=/127.0.0.1:9090 remote=/127.0.0.1:23549]
    4. read handler
    5. server read data from java.nio.channels.SocketChannel[connected local=/127.0.0.1:9090 remote=/127.0.0.1:23536], data is :client1
    6.  
    7. forward msg to java.nio.channels.SocketChannel[connected local=/127.0.0.1:9090 remote=/127.0.0.1:23549]
    8. java.nio.channels.SocketChannel[connected local=/127.0.0.1:9090 remote=/127.0.0.1:23536] : no data
    9. read handler
    10. server read data from java.nio.channels.SocketChannel[connected local=/127.0.0.1:9090 remote=/127.0.0.1:23549], data is :client2
    11.  
    12. forward msg to java.nio.channels.SocketChannel[connected local=/127.0.0.1:9090 remote=/127.0.0.1:23536]
    13. java.nio.channels.SocketChannel[connected local=/127.0.0.1:9090 remote=/127.0.0.1:23549] : no data

    客户端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: client2

    客户端2日志：

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

    测试：客户端1 正常退出关闭

    客户端1日志：

    1. exit
    2. client receive data from consolejava.io.BufferedInputStream@65231a33 : exit
    3.  
    4. client close :Socket[addr=localhost/127.0.0.1,port=9090,localport=23536]
    5. Exception in thread "main" java.net.SocketException: Socket closed
    6. 	at java.base/sun.nio.ch.NioSocketImpl.endRead(NioSocketImpl.java:248)
    7. 	at java.base/sun.nio.ch.NioSocketImpl.implRead(NioSocketImpl.java:327)
    8. 	at java.base/sun.nio.ch.NioSocketImpl.read(NioSocketImpl.java:350)
    9. 	at java.base/sun.nio.ch.NioSocketImpl$1.read(NioSocketImpl.java:803)
    10. 	at java.base/java.net.Socket$SocketInputStream.read(Socket.java:966)
    11. 	at java.base/java.io.InputStream.read(InputStream.java:218)
    12. 	at com.huawei.io.chatroom.bio.ChatClient.main(ChatClient.java:29)
    13.  
    14. Process finished with exit code 1

    服务端日志：

    1. read handler
    2. java.nio.channels.SocketChannel[connected local=/127.0.0.1:9090 remote=/127.0.0.1:23536] close

    测试：客户端2异常断开

    服务端日志：

    1. read handler
    2. java.nio.channels.SocketChannel[connected local=/127.0.0.1:9090 remote=/127.0.0.1:23549] disconnect
    3. java.net.SocketException: Connection reset
    4. 	at java.base/sun.nio.ch.SocketChannelImpl.throwConnectionReset(SocketChannelImpl.java:394)
    5. 	at java.base/sun.nio.ch.SocketChannelImpl.read(SocketChannelImpl.java:426)
    6. 	at com.huawei.io.chatroom.nio.sel.NIOSelectorDirectWriteServer.readHandler(NIOSelectorDirectWriteServer.java:107)
    7. 	at com.huawei.io.chatroom.nio.sel.NIOSelectorDirectWriteServer.doService(NIOSelectorDirectWriteServer.java:56)
    8. 	at com.huawei.io.chatroom.nio.sel.NIOSelectorDirectWriteServer.start(NIOSelectorDirectWriteServer.java:34)
    9. 	at com.huawei.io.chatroom.nio.sel.NIOSelectorDirectWriteServer.main(NIOSelectorDirectWriteServer.java:28)

    二、注册OP_WRITE写数据

  • 相关阅读:
    架构风格相关内容
    页面分配策略（驻留集、页面分配、置换策略、抖动现象、工作集）
    进程同步与互斥
    Android 的整体架构
    java-php-net-python-工会管理系统计算机毕业设计程序
    Apache POC(对Excel文件操作)
    SaaSBase：什么是天润融通？
    网关BL100做主站采集从机设备上云平台示例
    C++智能指针
    分享im即时通讯开发之WebSocket：概念、原理、易错常识
  • 原文地址：https://blog.csdn.net/u013771019/article/details/138153944