Netty——NIO（Selector处理read事件 消息边界问题）代码示例
Posted 小志的博客
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Netty——NIO（Selector处理read事件 消息边界问题）代码示例相关的知识，希望对你有一定的参考价值。
目录
一、处理消息边界的方式
- 第一种思路：固定消息长度，数据包大小一样，服务器按预定长度读取，缺点是浪费带宽。
- 第二种思路：按分隔符拆分，缺点是效率低。
- 第二种思路：TLV 格式，即 Type 类型、Length 长度、Value 数据，类型和长度已知的情况下，就可以方便获取消息大小，分配合适的 buffer，缺点是 buffer 需要提前分配，如果内容过大，则影响 server 吞吐量。
（1）、 Http 1.1 是 TLV 格式
（2）、Http 2.0 是 LTV 格式
二、未处理消息边界的代码示例
2.1、服务端代码示例
-
服务端代码
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<SelectionKey> 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();
2.2、客户端代码示例
-
客户端代码示例
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...");
2.3、工具类
-
工具类，打印输入、输出数据使用
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 < 256; i++) HEXDUMP_TABLE[i << 1] = DIGITS[i >>> 4 & 0x0F]; HEXDUMP_TABLE[(i << 1) + 1] = DIGITS[i & 0x0F]; int i; // Generate the lookup table for hex dump paddings for (i = 0; i < HEXPADDING.length; i++) int padding = HEXPADDING.length - i; StringBuilder buf = new StringBuilder(padding * 3); for (int j = 0; j < padding; j++) buf.append(" "); HEXPADDING[i] = buf.toString(); // Generate the lookup table for the start-offset header in each row (up to 64KiB). for (i = 0; i < HEXDUMP_ROWPREFIXES.length; i++) StringBuilder buf = new StringBuilder(12); buf.append(NEWLINE); buf.append(Long.toHexString(i << 4 & 0xFFFFFFFFL | 0x100000000L)); buf.setCharAt(buf.length() - 9, '|'); buf.append('|'); HEXDUMP_ROWPREFIXES[i] = buf.toString(); // Generate the lookup table for byte-to-hex-dump conversion for (i = 0; i < BYTE2HEX.length; i++) BYTE2HEX[i] = ' ' + StringUtil.byteToHexStringPadded(i); // Generate the lookup table for byte dump paddings for (i = 0; i < BYTEPADDING.length; i++) int padding = BYTEPADDING.length - i; StringBuilder buf = new StringBuilder(padding); for (int j = 0; j < padding; j++) buf.append(' '); BYTEPADDING[i] = buf.toString(); // Generate the lookup table for byte-to-char conversion for (i = 0; i < BYTE2CHAR.length; i++) if (i <= 0x1f || i >= 0x7f) BYTE2CHAR[i] = '.'; else BYTE2CHAR[i] = (char) i; /** * 打印所有内容 * @param buffer */ public static void debugAll(ByteBuffer buffer) int oldlimit = buffer.limit(); buffer.limit(buffer.capacity()); StringBuilder origin = new StringBuilder(256); appendPrettyHexDump(origin, buffer, 0, buffer.capacity()); System.out.println("+--------+-------------------- all ------------------------+----------------+"); System.out.printf("position: [%d], limit: [%d]\\n", buffer.position(), oldlimit); System.out.println(origin); buffer.limit(oldlimit); /** * 打印可读取内容 * @param buffer */ public static void debugRead(ByteBuffer buffer) StringBuilder builder = new StringBuilder(256); appendPrettyHexDump(builder, buffer, buffer.position(), buffer.limit() - buffer.position()); System.out.println("+--------+-------------------- read -----------------------+----------------+"); System.out.printf("position: [%d], limit: [%d]\\n", buffer.position(), buffer.limit()); System.out.println(builder); public static void main(String[] args) ByteBuffer buffer = ByteBuffer.allocate(10); buffer.put(new byte[]97, 98, 99, 100); debugAll(buffer); private static void appendPrettyHexDump(StringBuilder dump, ByteBuffer buf, int offset, int length) if (isOutOfBounds(offset, length, buf.capacity())) throw new IndexOutOfBoundsException( "expected: " + "0 <= offset(" + offset + ") <= offset + length(" + length + ") <= " + "buf.capacity(" + buf.capacity() + ')'); if (length == 0) return; dump.append( " +-------------------------------------------------+" + NEWLINE + " | 0 1 2 3 4 5 6 7 8 9 a b c d e f |" + NEWLINE + "+--------+-------------------------------------------------+----------------+"); final int startIndex = offset; final int fullRows = length >>> 4; final int remainder = length & 0xF; // Dump the rows which have 16 bytes. for (int row = 0; row < fullRows; row++) int rowStartIndex = (row << 4) + startIndex; // Per-row prefix. appendHexDumpRowPrefix(dump, row, rowStartIndex); // Hex dump int rowEndIndex = rowStartIndex + 16; for (int j = rowStartIndex; j < rowEndIndex; j++) dump.append(BYTE2HEX[getUnsignedByte(buf, j)]); dump.append(" |"); // ASCII dump for (int j = rowStartIndex; j < rowEndIndex; j++) dump.append(BYTE2CHAR[getUnsignedByte(buf, j)]); dump.append('|'); // Dump the last row which has less than 16 bytes. if (remainder != 0) int rowStartIndex = (fullRows << 4) + startIndex; appendHexDumpRowPrefix(dump, fullRows, rowStartIndex); // Hex dump int rowEndIndex = rowStartIndex + remainder; for (int j = rowStartIndex; j < rowEndIndex; j++) dump.append(BYTE2HEX[getUnsignedByte(buf, j)]); dump.append(HEXPADDING[remainder]); dump.app以上是关于Netty——NIO（Selector处理read事件 消息边界问题）代码示例的主要内容，如果未能解决你的问题，请参考以下文章
Netty——网络编程 NIO（Selector处理read事件）代码示例
Netty——NIO（Selector处理read事件 消息边界问题）代码示例
Netty——NIO（Selector处理read事件 消息边界问题）代码示例
Netty——网络编程 NIO（Selector处理read事件 消息边界问题）代码示例