Netty——网络编程 NIO（Selector处理read事件 消息边界问题）代码示例

服务端代码

package com.example.nettytest.nio.day3;

import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;
import java.util.Iterator;

import static com.example.nettytest.nio.day1.ByteBufferUtil.debugAll;

/**
 * @description: Selector处理read事件（消息边界问题） 代码示例
 * @author: xz
 * @create: 2022-09-04
 */
@Slf4j
public class Test5Server {
    public static void main(String[] args) {
        nioSelectorReadServer();
    }
    /**
     * 1、消息边界问题
     * */
    private static void nioSelectorReadServer(){
        try {
            // 1. 创建 selector, 管理多个 channel
            Selector selector = Selector.open();
            ServerSocketChannel ssc = ServerSocketChannel.open();
            ssc.configureBlocking(false);
            // 2. 建立 selector 和 channel 的联系（注册）
            // SelectionKey 就是将来事件发生后，通过它可以知道事件和哪个channel的事件
            SelectionKey sscKey = ssc.register(selector, 0, null);
            // key 只关注 accept 事件
            sscKey.interestOps(SelectionKey.OP_ACCEPT);
            log.debug("sscKey:{}", sscKey);
            ssc.bind(new InetSocketAddress(8080));
            while (true) {
                // 3. select 方法, 没有事件发生，线程阻塞，有事件，线程才会恢复运行
                // select 在事件未处理时，它不会阻塞, 事件发生后要么处理，要么取消，不能置之不理
                selector.select();
                // 4. 处理事件, selectedKeys 内部包含了所有发生的事件
                Iterator iter = selector.selectedKeys().iterator(); // accept, read
                while (iter.hasNext()) {
                    SelectionKey key = iter.next();
                    // 处理key 时，要从 selectedKeys 集合中删除，否则下次处理就会有问题
                    iter.remove();
                    log.debug("key: {}", key);
                    // 5. 区分事件类型
                    if (key.isAcceptable()) { // 如果是 accept
                        ServerSocketChannel channel = (ServerSocketChannel) key.channel();
                        SocketChannel sc = channel.accept();
                        sc.configureBlocking(false);
                        SelectionKey scKey = sc.register(selector, 0, null);
                        scKey.interestOps(SelectionKey.OP_READ);
                        log.debug("{}", sc);
                        log.debug("scKey:{}", scKey);
                    } else if (key.isReadable()) { // 如果是 read
                        try {
                            SocketChannel channel = (SocketChannel) key.channel(); // 拿到触发事件的channel
                            //分配ByteBuffer容量4个字节
                            ByteBuffer buffer = ByteBuffer.allocate(4);
                            int read=channel.read(buffer); // 如果是正常断开，read 的方法的返回值是 -1
                            if(read == -1) {
                                key.cancel();
                            } else {
                                buffer.flip();
                                System.out.println(Charset.defaultCharset().decode(buffer));
                            }
                        } catch (IOException e) {
                            e.printStackTrace();
                            // 因为客户端断开了,因此需要将 key 取消（从 selector 的 keys 集合中真正删除 key）
                            key.cancel();
                        }

                    }
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

客户端代码示例

package com.example.nettytest.nio.day3;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;

/**
 * @description:
 * @author: xz
 * @create: 2022-09-04
 */
public class Test5Client {
    public static void main(String[] args) throws IOException {
        SocketChannel sc = SocketChannel.open();
        sc.connect(new InetSocketAddress("localhost", 8080));
        SocketAddress address = sc.getLocalAddress();
        //超过服务端设置的ByteBuffer容量4个字节
        sc.write(Charset.defaultCharset().encode("123456789abcd\n"));
        System.out.println("waiting...");
    }
}

工具类，打印输入、输出数据使用

package com.example.nettytest.nio.day1;

import io.netty.util.internal.StringUtil;

import java.nio.ByteBuffer;

import static io.netty.util.internal.MathUtil.isOutOfBounds;
import static io.netty.util.internal.StringUtil.NEWLINE;

public class ByteBufferUtil {
    private static final char[] BYTE2CHAR = new char[256];
    private static final char[] HEXDUMP_TABLE = new char[256 * 4];
    private static final String[] HEXPADDING = new String[16];
    private static final String[] HEXDUMP_ROWPREFIXES = new String[65536 >>> 4];
    private static final String[] BYTE2HEX = new String[256];
    private static final String[] BYTEPADDING = new String[16];

    static {
        final char[] DIGITS = "0123456789abcdef".toCharArray();
        for (int i = 0; i  4 & 0x0F];
            HEXDUMP_TABLE[(i