码迷,mamicode.com
首页 > Web开发 > 详细

Netty连接处理那些事

时间:2019-11-12 00:26:35      阅读:125      评论:0      收藏:0      [点我收藏+]

标签:oca   责任   分析   等于   之一   责任链模式   safe   complete   面试   

编者注:Netty是Java领域有名的开源网络库,特点是高性能和高扩展性,因此很多流行的框架都是基于它来构建的,比如我们熟知的Dubbo、Rocketmq、Hadoop等,针对高性能RPC,一般都是基于Netty来构建,比如soft-bolt。总之一句话,Java小伙伴们需要且有必要学会使用Netty并理解其实现原理。
关于Netty的入门讲解可参考:Netty 入门,这一篇文章就够了

Netty的连接处理就是IO事件的处理,IO事件包括读事件、ACCEPT事件、写事件和OP_CONNECT事件。

IO事件的处理是结合ChanelPipeline来做的,一个IO事件到来,首先进行数据的读写操作,然后交给ChannelPipeline进行后续处理,ChannelPipeline中包含了channelHandler链(head + 自定义channelHandler + tail)。
使用channelPipeline和channelHandler机制,起到了解耦和可扩展的作用。一个IO事件的处理,包含了多个处理流程,这些处理流程正好对应channelPipeline中的channelHandler。如果对数据处理有新的需求,那么就新增channelHandler添加到channelPipeline中,这样实现很6,以后自己写代码可以参考。

说到这里,一般为了满足扩展性要求,常用2种模式:

  • 方法模板模式:模板中定义了各个主流程,并且留下对应hook方法,便于扩展。
  • 责任链模式:串行模式,可以动态添加链数量和对应回调方法。

netty的channelHandlerchannelPipeline可以理解成就是责任链模式,通过动态增加channelHandler可达到复用和高扩展性目的。

了解netty连接处理机制之前需要了解下NioEventLoop模型,其中处理连接事件的架构图如下:

技术图片

对应的处理逻辑源码为:

// 处理各种IO事件
private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
    final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();

    try {
        int readyOps = k.readyOps();
        if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
            // OP_CONNECT事件,client连接上客户端时触发的事件
            int ops = k.interestOps();
            ops &= ~SelectionKey.OP_CONNECT;
            k.interestOps(ops);
            unsafe.finishConnect();
        }

        if ((readyOps & SelectionKey.OP_WRITE) != 0) {
            ch.unsafe().forceFlush();
        }

        if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
            // 注意,这里读事件和ACCEPT事件对应的unsafe实例是不一样的
            // 读事件 -> NioByteUnsafe,  ACCEPT事件 -> NioMessageUnsafe
            unsafe.read();
        }
    } catch (CancelledKeyException ignored) {
        unsafe.close(unsafe.voidPromise());
    }
}

从上面代码来看,事件主要分为3种,分别是OP_CONNECT事件、写事件和读事件(也包括ACCEPT事件)。下面分为3部分展开:

ACCEPT事件

// NioMessageUnsafe
public void read() {
    assert eventLoop().inEventLoop();
    final ChannelConfig config = config();
    final ChannelPipeline pipeline = pipeline();
    final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle();
    allocHandle.reset(config);
 
    boolean closed = false;
    Throwable exception = null;
    try {
        do {
            // 调用java socket的accept方法,接收请求
            int localRead = doReadMessages(readBuf);
            // 增加统计计数
            allocHandle.incMessagesRead(localRead);
        } while (allocHandle.continueReading());
    } catch (Throwable t) {
        exception = t;
    }
 
    // readBuf中存的是NioChannel
    int size = readBuf.size();
    for (int i = 0; i < size; i ++) {
        readPending = false;
        // 触发fireChannelRead
        pipeline.fireChannelRead(readBuf.get(i));
    }
    readBuf.clear();
    allocHandle.readComplete();
    pipeline.fireChannelReadComplete();
}

连接建立好之后就该连接的channel注册到workGroup中某个NIOEventLoop的selector中,注册操作是在fireChannelRead中完成的,这一块逻辑就在ServerBootstrapAcceptor.channelRead中。

// ServerBootstrapAcceptor
public void channelRead(ChannelHandlerContext ctx, Object msg) {
    final Channel child = (Channel) msg;
 
    // 设置channel的pipeline handler,及channel属性
    child.pipeline().addLast(childHandler);
    setChannelOptions(child, childOptions, logger);
 
    for (Entry<AttributeKey<?>, Object> e: childAttrs) {
        child.attr((AttributeKey<Object>) e.getKey()).set(e.getValue());
    }
 
    try {
        // 将channel注册到childGroup中的Selector上
        childGroup.register(child).addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                if (!future.isSuccess()) {
                    forceClose(child, future.cause());
                }
            }
        });
    } catch (Throwable t) {
        forceClose(child, t);
    }
}

READ事件

// NioByteUnsafe
public final void read() {
    final ChannelConfig config = config();
    final ChannelPipeline pipeline = pipeline();
    final ByteBufAllocator allocator = config.getAllocator();
    final RecvByteBufAllocator.Handle allocHandle = recvBufAllocHandle();
    allocHandle.reset(config);
 
    ByteBuf byteBuf = null;
    boolean close = false;
    try {
        do {
            byteBuf = allocHandle.allocate(allocator);
            // 从channel中读取数据,存放到byteBuf中
            allocHandle.lastBytesRead(doReadBytes(byteBuf));
 
            allocHandle.incMessagesRead(1);
            readPending = false;
 
            // 触发fireChannelRead
            pipeline.fireChannelRead(byteBuf);
            byteBuf = null;
        } while (allocHandle.continueReading());

        // 触发fireChannelReadComplete,如果在fireChannelReadComplete中执行了ChannelHandlerContext.flush,则响应结果返回给客户端
        allocHandle.readComplete();
        // 触发fireChannelReadComplete
        pipeline.fireChannelReadComplete();
 
        if (close) {
            closeOnRead(pipeline);
        }
    } catch (Throwable t) {
        if (!readPending && !config.isAutoRead()) {
            removeReadOp();
        }
    }
}

写事件

正常情况下一般是不会注册写事件的,如果Socket发送缓冲区中没有空闲内存时,再写入会导致阻塞,此时可以注册写事件,当有空闲内存(或者可用字节数大于等于其低水位标记)时,再响应写事件,并触发对应回调。

if ((readyOps & SelectionKey.OP_WRITE) != 0) {
    // 写事件,从flush操作来看,虽然之前没有向socket缓冲区写数据,但是已经写入到
    // 了chnanel的outboundBuffer中,flush操作是将数据从outboundBuffer写入到
    // socket缓冲区
    ch.unsafe().forceFlush();
}

CONNECT事件

该事件是client触发的,由主动建立连接这一侧触发的。

if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
    // OP_CONNECT事件,client连接上客户端时触发的事件
    int ops = k.interestOps();
    ops &= ~SelectionKey.OP_CONNECT;
    k.interestOps(ops);
 
    // 触发finishConnect事件,其中就包括fireChannelActive事件,如果有自定义的handler有channelActive方法,则会触发
    unsafe.finishConnect();
}

推荐阅读

欢迎小伙伴关注【TopCoder】阅读更多精彩好文。
技术图片

Netty连接处理那些事

标签:oca   责任   分析   等于   之一   责任链模式   safe   complete   面试   

原文地址:https://www.cnblogs.com/luoxn28/p/11839273.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!