以上三篇内容主要讲了NIO的三大组件
、ByteBuffer
、文件编程
、阻塞
、非阻塞
、 Selector
等,需要了解像详情的请移步查看。
本章主要讲解如何处理在消息传递过程中的边界问题。
如图所示:在实际项目中,消息有可能要比ByteBuffer长,或者比ByteBuffer短; 针对以上的几种情况,应该如何去处理呢?有两种方案:
上代码(⚠️一定要注意代码中的注释):
@Slf4j
public class Server {
private static void split(ByteBuffer source) {
source.flip();
for (int i = 0; i < source.limit(); i++) {
// 找到一条完整消息
if (source.get(i) == '\n') {
int length = i + 1 - source.position();
// 把这条完整消息存入新的 ByteBuffer
ByteBuffer target = ByteBuffer.allocate(length);
// 从 source 读,向 target 写
for (int j = 0; j < length; j++) {
target.put(source.get());
}
debugAll(target);
}
}
source.compact();
}
public static void main(String[] args) throws IOException {
// 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);
ByteBuffer buffer = ByteBuffer.allocate(16) // 利用 附件的方式将buffer注册关联到selectionKey上
SelectionKey scKey = sc.register(selector, 0, buffer);
scKey.interestOps(SelectionKey.OP_READ);
log.debug("{}", sc);
log.debug("scKey:{}", scKey);
} else if (key.isReadable()) { // 如果是 read
try {
SocketChannel channel = (SocketChannel) key.channel(); // 拿到触发事件的channel
// 取上次注册的buffer
ByteBuffer buffer = (ByteBuffer)key.attachment();
int read = channel.read(buffer); // 如果是正常断开,read 的方法的返回值是 -1
if(read == -1) {
key.cancel();
} else {
split(buffer)
// 这里说明原有的buffer满了
if (buffer.position() == buffer.limit()) {
ByteBuffer newByteBuffer = ByteBuffer.allocate(buffer.capacity() * 2 )
buffer.flip();
newByteBuffer.put(buffer);
// 替换原有的附件buffer
key.attach(newByteBuffer);
}
}
} catch (IOException e) {
e.printStackTrace();
key.cancel(); // 因为客户端断开了,因此需要将 key 取消(从 selector 的 keys 集合中真正删除 key)
}
}
}
}
}
}
public class Client {
public static void main(String[] args) throws IOException {
SocketChannel sc = SocketChannel.open();
sc.connect(new InetSocketAddress("localhost", 8080));
SocketAddress address = sc.getLocalAddress();
sc.write(Chaset.defaultCharset().encode("012345677890abcdef"))
System.in.read();
}
}
demo 解析: 上述server端代码用到了附件
这个概念
ByteBuffer buffer = ByteBuffer.allocate(16) // 利用 附>件的方式将buffer注册关联到selectionKey上
SelectionKey scKey = sc.register(selector, 0, buffer);
重新关联附件buffer key.attach(newByteBuffer);
上述代码就是简单做了一个消息边界的处理,相信大家也看到了一些问题,它只能是做到自动扩容,无法自适应,也就是缩小。暂时先提前告诉大家Netty是可以做到自适应的。
如何处理消息边界问题以及ByteBuffer大小分配的问题已经说完了,接下来给大家说一下ByteBuffer的大小如何分配
的注意点。
每个Channel都需要记录可能被切分的消息,因为ByteBuffer不能够被多个Channel共同使用,因此需要为每个channel维护一个独立的ByteBUffer
好了本次的文章就到这里了后续再为大家带来关于Netty的更多内容。切记:一定要好好消化上述的demo案例。
扫码关注腾讯云开发者
领取腾讯云代金券
Copyright © 2013 - 2025 Tencent Cloud. All Rights Reserved. 腾讯云 版权所有
深圳市腾讯计算机系统有限公司 ICP备案/许可证号:粤B2-20090059 深公网安备号 44030502008569
腾讯云计算(北京)有限责任公司 京ICP证150476号 | 京ICP备11018762号 | 京公网安备号11010802020287
Copyright © 2013 - 2025 Tencent Cloud.
All Rights Reserved. 腾讯云 版权所有