Netty——NIO（Selector处理read事件）代码示例
目录
- 一、服务端代码示例
- 二、客户端代码示例
- 三、工具类
- 四、启动一个客户端测试
- 五、启动2个客户端测试
- 六、服务端示例代码中为何要使用 iter.remove()
- 七、服务端示例代码中cancel()的作用
一、服务端代码示例
-
服务端代码示例
package com.example.nettytest.nio.day3; import com.example.nettytest.nio.day1.ByteBufferUtil; 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.util.Iterator; import java.util.Set; /** * @description: Selector处理read事件 代码示例 * @author: xz * @create: 2022-09-01 */ @Slf4j public class Test3Server { public static void main(String[] args) { nioSelectorReadServer(); } /** * Selector处理read事件 方法 * */ private static void nioSelectorReadServer(){ try (ServerSocketChannel channel = ServerSocketChannel.open()) {//1、创建服务器 //绑定监听端口 channel.bind(new InetSocketAddress(8080)); //设置为非阻塞模式 channel.configureBlocking(false); System.out.println(channel); /** * 2、创建Selector，管理多个channel */ Selector selector = Selector.open(); /** * 3、建立Selector与channel的联系（也称为注册） * key只关注事件,绑定的事件类型4种，connect - 客户端连接成功时触发；accept - 服务器端成功接受连接时触发; read - 数据可读入时触发;write - 数据可写出时触发， * */ channel.register(selector, SelectionKey.OP_ACCEPT); while (true) { /** * 4、select 方法, 没有事件发生，线程阻塞，有事件，线程才会恢复运行 * select 在事件未处理时，它不会阻塞 */ int count = selector.select(); log.debug("select count: {}", count); // 获取所有事件 Set<SelectionKey> keys = selector.selectedKeys(); // 遍历所有事件，逐一处理 Iterator<SelectionKey> iter = keys.iterator(); while (iter.hasNext()) { SelectionKey key = iter.next(); // 判断事件类型 如果事件类型是 accept - 服务器端成功接受连接时触发; if (key.isAcceptable()) { //channel中有哪些key ServerSocketChannel c = (ServerSocketChannel) key.channel(); // 必须处理 selector 上的 channel SocketChannel sc = c.accept(); //设置为非阻塞模式 sc.configureBlocking(false); //建立Selector与channel的联系（也称为注册） sc.register(selector, SelectionKey.OP_READ); log.debug("连接已建立: {}", sc); // 判断事件类型 如果事件类型是read - 数据可读入时触发 } else if (key.isReadable()) { //channel中有哪些key SocketChannel sc = (SocketChannel) key.channel(); //申请ByteBuffer容量 ByteBuffer buffer = ByteBuffer.allocate(128); //从channel中读并写入到buffer int read = sc.read(buffer); if(read == -1) { //如果读到末尾 //取消注册在 selector 上的 channel key.cancel(); //关闭channel sc.close(); } else { //如果没有读到末尾 //切换到读模式 buffer.flip(); //打印可读取内容 ByteBufferUtil.debugRead(buffer); } } // 处理完毕，必须将事件移除 iter.remove(); } } } catch (IOException e) { e.printStackTrace(); } } }
二、客户端代码示例
-
客户端代码示例
package com.example.nettytest.nio.day3; import java.io.IOException; import java.net.Socket; /** * @description: * @author: xz * @create: 2022-09-01 */ public class Test3Client { public static void main(String[] args) { try (Socket socket = new Socket("localhost", 8080)) { System.out.println(socket); socket.getOutputStream().write("hello world".getBytes()); System.in.read(); } catch (IOException e) { e.printStackTrace(); } } }
三、工具类
-
工具类，打印输入、输出数据使用
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.append(" |"); // Ascii dump for (int j = rowStartIndex; j < rowEndIndex; j++) { dump.append(BYTE2CHAR[getUnsignedByte(buf, j)]); } dump.append(BYTEPADDING[remainder]); dump.append('|'); } dump.append(NEWLINE + "+--------+-------------------------------------------------+----------------+"); } private static void appendHexDumpRowPrefix(StringBuilder dump, int row, int rowStartIndex) { if (row < HEXDUMP_ROWPREFIXES.length) { dump.append(HEXDUMP_ROWPREFIXES[row]); } else { dump.append(NEWLINE); dump.append(Long.toHexString(rowStartIndex & 0xFFFFFFFFL | 0x100000000L)); dump.setCharAt(dump.length() - 9, '|'); dump.append('|'); } } public static short getUnsignedByte(ByteBuffer buffer, int index) { return (short) (buffer.get(index) & 0xFF); } }
四、启动一个客户端测试
- 先启动服务端
由下图输出结果可知，由于没有事件发生，再select 方法处，线程阻塞。
- 再启动客户端
- 再查看服务端控制台输出
由下图输出结果可知，由于有事件发生，再select 方法处，线程会恢复运行。
五、启动2个客户端测试
- 再启动一个客户端
- 再次查看服务端控制台输出
由下图输出结果可知，由于有事件发生，再select 方法处，线程再次恢复运行。
六、服务端示例代码中为何要使用 iter.remove()
- 因为 select 在事件发生后，就会将相关的 key 放入 selectedKeys 集合，但不会在处理完后从 selectedKeys 集合中移除，需要我们自己编码删除。
七、服务端示例代码中cancel()的作用
- cancel() 会取消注册在 selector 上的 channel，并从 keys 集合中删除 key 后续不会再监听事件。