欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页

Netty中的ChannelPipeline及其实现

程序员文章站 2022-05-08 17:56:51
...

下面是ChannelPipeline的java doc文档的部分介绍

/**
 * A list of {@link ChannelHandler}s which handles or intercepts inbound events and outbound operations of a
 * {@link Channel}.  {@link ChannelPipeline} implements an advanced form of the
 * 是一个处理或者在一个channel上的拦截入站事件和出站事件操作的ChannelHandler的集合
 * <a href="http://www.oracle.com/technetwork/java/interceptingfilter-142169.html">Intercepting Filter</a> pattern
 * to give a user full control over how an event is handled and how the {@link ChannelHandler}s in a pipeline
 * interact with each other.
 * channelpipeline实现了一种高级的拦截过滤器模式,让用户完全自主控制一个事件是怎么被处理的,以及在pipeline中channelHandler是怎么跟彼此交互的
 * <h3>Creation of a pipeline</h3>
 * 创建一个pipeline
 * Each channel has its own pipeline and it is created automatically when a new channel is created.
 * 每个channel都会有自己的pipeline,它是在一个新的channel被创建的时候自动创建的
 * <h3>How an event flows in a pipeline</h3>
 * 一个事件是怎么在一个pipeline中流动 的
 * The following diagram describes how I/O events are processed by {@link ChannelHandler}s in a {@link ChannelPipeline}
 * typically. An I/O event is handled by either a {@link ChannelInboundHandler} or a {@link ChannelOutboundHandler}
 * and be forwarded to its closest handler by calling the event propagation methods defined in
 * {@link ChannelHandlerContext}, such as {@link ChannelHandlerContext#fireChannelRead(Object)} and
 * {@link ChannelHandlerContext#write(Object)}.
 * 下面的类图描绘了IO事件是怎样被在channelpipeline中的channelhandler所处理的。
 * 一个io事件被一个ChannelInboundHandler或者一个ChannelOutboundHandler所处理,
   通过在ChannelHandlerContext中定义的事件传播机制去调用离他们最近的handler,
 * 例如ChannelHandlerContext#fireChannelRead(Object) 和 ChannelHandlerContext#write(Object)
 * <pre>
 *                                                 I/O Request
 *                                            via {@link Channel} or
 *                                        {@link ChannelHandlerContext}
 *                                                      |
 *  +---------------------------------------------------+---------------+
 *  |                           ChannelPipeline         |               |
 *  |                                                  \|/              |
 *  |    +---------------------+            +-----------+----------+    |
 *  |    | Inbound Handler  N  |            | Outbound Handler  1  |    |
 *  |    +----------+----------+            +-----------+----------+    |
 *  |              /|\                                  |               |
 *  |               |                                  \|/              |
 *  |    +----------+----------+            +-----------+----------+    |
 *  |    | Inbound Handler N-1 |            | Outbound Handler  2  |    |
 *  |    +----------+----------+            +-----------+----------+    |
 *  |              /|\                                  .               |
 *  |               .                                   .               |
 *  | ChannelHandlerContext.fireIN_EVT() ChannelHandlerContext.OUT_EVT()|
 *  |        [ method call]                       [method call]         |
 *  |               .                                   .               |
 *  |               .                                  \|/              |
 *  |    +----------+----------+            +-----------+----------+    |
 *  |    | Inbound Handler  2  |            | Outbound Handler M-1 |    |
 *  |    +----------+----------+            +-----------+----------+    |
 *  |              /|\                                  |               |
 *  |               |                                  \|/              |
 *  |    +----------+----------+            +-----------+----------+    |
 *  |    | Inbound Handler  1  |            | Outbound Handler  M  |    |
 *  |    +----------+----------+            +-----------+----------+    |
 *  |              /|\                                  |               |
 *  +---------------+-----------------------------------+---------------+
 *                  |                                  \|/
 *  +---------------+-----------------------------------+---------------+
 *  |               |                                   |               |
 *  |       [ Socket.read() ]                    [ Socket.write() ]     |
 *  |                                                                   |
 *  |  Netty Internal I/O Threads (Transport Implementation)            |
 *  +-------------------------------------------------------------------+
 * </pre>
 * An inbound event is handled by the inbound handlers in the bottom-up direction as shown on the left side of the
 * diagram.  An inbound handler usually handles the inbound data generated by the I/O thread on the bottom of the
 * diagram.  The inbound data is often read from a remote peer via the actual input operation such as
 * {@link SocketChannel#read(ByteBuffer)}.  If an inbound event goes beyond the top inbound handler, it is discarded
 * silently, or logged if it needs your attention.
 * 一个入站事件是被入站handlers所处理,在这个类图的左边显示的那样,以一个从下至上的方向进行,
   一个入站handler通常处理通过在类图底部的io线程生成的入站数据。
 * 入站数据经常从一个远程节点读取,通过一个真是存在的输入操作,如果一个入站事件从顶部的入站handler走出去了,
   它就会被默默的丢弃掉,或者如果你需要关注他,你通过日志的方式记录下来。
 * <p>
 * An outbound event is handled by the outbound handler in the top-down direction as shown on the right side of the
 * diagram.  An outbound handler usually generates or transforms the outbound traffic such as write requests.
 * If an outbound event goes beyond the bottom outbound handler, it is handled by an I/O thread associated with the
 * {@link Channel}. The I/O thread often performs the actual output operation such as
 * {@link SocketChannel#write(ByteBuffer)}.
 * 如果一个出站的事件走出了底部的出站handler,他就会被与这个channel相关的io线程处理。
 * <p>
 *
 * <h3>Forwarding an event to the next handler</h3>
 * 传播一个事件到下一个handler中
 * As you might noticed in the diagram shows, a handler has to invoke the event propagation methods in
 * {@link ChannelHandlerContext} to forward an event to its next handler.  Those methods include:
 * <ul>
        一个handler必须调用ChannelHandlerContext中的时间传播方法来将一个事件传播到它的下一个handler中。比如ctx.fireChannelActive();
 * </pre>
 *
 * <h3>Building a pipeline</h3>
 *
 * // Tell the pipeline to run MyBusinessLogicHandler's event handler methods
 * // in a different thread than an I/O thread so that the I/O thread is not blocked by
 * // a time-consuming task.
 * // If your business logic is fully asynchronous or finished very quickly, you don't
 * // need to specify a group.
 * pipeline.addLast(group, "handler", new MyBusinessLogicHandler());
 * 告诉pipeline在一个与IO线程不同的线程中去运行MyBusinessLogicHandler's的事件处理方法,这样IO线程就不会因为一个耗时的任务而被阻塞
 * 如果你的业务逻辑是完全异步的 ,或者执行速度很快,你可以不必指定一个group.
 * </pre>
 *
 * <h3>Thread safety</h3>
 * <p>
 * A {@link ChannelHandler} can be added or removed at any time because a {@link ChannelPipeline} is thread safe.
 * For example, you can insert an encryption handler when sensitive information is about to be exchanged, and remove it
 * after the exchange.
 * channelHandler可以在任何时候进行添加和移除,因为ChannelPipeline是线程安全的。
 * 例如,当敏感的信息要进行交换时,你可以插入一个加密的handler,当信息交换完毕之后,删除掉这个handler.
 */

我们主要看Pipeline的默认实现DefaultChannelPipeline中的addLast方法的实现

   @Override
    public final ChannelPipeline addLast(EventExecutorGroup group, String name, ChannelHandler handler) {
        final AbstractChannelHandlerContext newCtx;
        synchronized (this) {
            checkMultiplicity(handler);

            //创建一个新的handler上下文
        newCtx = newContext(group, filterName(name, handler), handler);
        //将这个新创建的handler上下文添加到pipeline的末端,这里面使用链表的操作方式。
        addLast0(newCtx);

        // If the registered is false it means that the channel was not registered on an eventloop yet.
        // In this case we add the context to the pipeline and add a task that will call
        // ChannelHandler.handlerAdded(...) once the channel is registered.
        //如果registered是fasle,这就意味着channel目前还没有被注册到eventloop中,
        //在这种情况下,我们会将context添加到这个pipeline中,并且添加一个一旦channel被注册了就会调用ChannelHandler.handlerAdded()的任务
        //很明显,一开始我们是没有注册的,所以为false
        if (!registered) {
            newCtx.setAddPending();
            callHandlerCallbackLater(newCtx, true);
            return this;
        }

        EventExecutor executor = newCtx.executor();
        if (!executor.inEventLoop()) {
            newCtx.setAddPending();
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    callHandlerAdded0(newCtx);
                }
            });
            return this;
        }
    }
    callHandlerAdded0(newCtx);
    return this;
}
    private void callHandlerCallbackLater(AbstractChannelHandlerContext ctx, boolean added) {
        assert !registered;
        //added为true,实例化一个PendingHandlerAddedTask,这个类是一个Runnable接口的具体实现,是一个线程任务,
        PendingHandlerCallback task = added ? new PendingHandlerAddedTask(ctx) : new PendingHandlerRemovedTask(ctx);
        PendingHandlerCallback pending = pendingHandlerCallbackHead;
        //pending为空的时候,说明任务还没有,那么将这个任务赋予pendingHandlerCallbackHead,作为头一个任务
        //如果有,则赋给头一个任务的下一个(链表操作)
        if (pending == null) {
            pendingHandlerCallbackHead = task;
        } else {
            // Find the tail of the linked-list.
            while (pending.next != null) {
                pending = pending.next;
            }
            pending.next = task;
        }
    }

将handler添加到链表的最后的方法实现

    private void addLast0(AbstractChannelHandlerContext newCtx) {
        AbstractChannelHandlerContext prev = tail.prev;
        newCtx.prev = prev;
        newCtx.next = tail;
        prev.next = newCtx;
        tail.prev = newCtx;
    }
最终的目的就是让新创建的newCtx插入到tail之前,并且更新newCtx前后两个ctx的next以及prev的值,以及本身的next和prev的值,实现链表的插入

相关标签: pipeline