Pipeline的创建

    NioServerSocketChannel.java

    1. public static void main(String[] args) throws InterruptedException {
    2.     // parentGroup中的eventLoop将来是用于与处理客户端连接请求的parentChannel进行绑定的
    3.     EventLoopGroup parentGroup = new NioEventLoopGroup();
    4.     // childGroup中的eventLoop将来是用于与处理客户端读写请求的childChannel进行绑定的
    5.     EventLoopGroup childGroup = new NioEventLoopGroup();
    6.     final EventLoopGroup someGroup = new NioEventLoopGroup();
    7.     try {
    8.         ServerBootstrap bootstrap = new ServerBootstrap();
    9.         bootstrap.group(parentGroup, childGroup)
    10.                 //从这里开始跟踪，跟踪NioServerSocketChannel.class
    11.                 .channel(NioServerSocketChannel.class)
    12.                 //这两个方法都是用于设置Channel属性的
    13.                 //attr的方法一般是用于parentGroup中,一般会在当客户端与服务端连接建立的时候触发调用
    14.                  .attr(AttributeKey.valueOf("depart"), "市场部")
    15.                 //chaild的方法一般用于childGroup中，一般用于客户端往服务端发送消息的时候触发调用
    16.                  .childAttr(AttributeKey.valueOf("addr"), "北京海淀")
    17.                 .childHandler(new ChannelInitializer<SocketChannel>() {
    18.                     @Override
    19.                     protected void initChannel(SocketChannel ch) throws Exception {
    21.                         ChannelPipeline pipeline = ch.pipeline();
    22.                         pipeline.addLast(new StringDecoder());
    23.                         pipeline.addLast(new StringEncoder());
    24.                         pipeline.addLast(new SomeSocketServerHandler());
    26.                          Object depart = ch.attr(AttributeKey.valueOf("depart")).get();
    27.                          Object addr = ch.attr(AttributeKey.valueOf("addr")).get();
    28.                          System.out.println("depart = " + depart);
    29.                          System.out.println("addr = " + addr);
    31.                     }
    32.                 });
    33.         ChannelFuture future = bootstrap.bind(8888).sync();
    34.         System.out.println("服务器已启动。。。");
    35.         future.channel().closeFuture().sync();
    36.     } finally {
    37.         parentGroup.shutdownGracefully();
    38.         childGroup.shutdownGracefully();
    39.     }
    40. }

    NioServerSocketChannel.java

    1. public NioServerSocketChannel(ServerSocketChannel channel) {
    2.     // 第一个参数：父channel
    3.     // 第二个参数：Nio原生channel
    4.     // 第三个参数：指定其关注的事件为 接收客户端连接
    5.     // 跟踪super
    6.     super(null, channel, SelectionKey.OP_ACCEPT);
    7.     // config用于对当前channel进行配置的
    8.     config = new NioServerSocketChannelConfig(this, javaChannel().socket());
    9. }

    AbstractNioMessageChannel.java

    1. protected AbstractNioMessageChannel(Channel parent, SelectableChannel ch, int readInterestOp) {
    2.       // 跟踪super
    3.     super(parent, ch, readInterestOp);
    4. }

    AbstractNioChannel.java

    1. protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) {
    2.       //跟踪super
    3.     super(parent);
    4.     // NIO原生channel
    5.     this.ch = ch;
    6.     // 初始化其关注事件
    7.     this.readInterestOp = readInterestOp;
    8.     try {
    9.         // 指定这里使用的是非阻塞IO，即NIO
    10.         ch.configureBlocking(false);
    11.     } catch (IOException e) {
    12.         try {
    13.             ch.close();
    14.         } catch (IOException e2) {
    15.             if (logger.isWarnEnabled()) {
    16.                 logger.warn(
    17.                         "Failed to close a partially initialized socket.", e2);
    18.             }
    19.         }
    21.         throw new ChannelException("Failed to enter non-blocking mode.", e);
    22.     }
    23. }

    AbstractChannel.java

    1. //其Pipeline对象是在这里创建的
    2. protected AbstractChannel(Channel parent) {
    3.     this.parent = parent;
    4.     // 生成channel的id
    5.     id = newId();
    6.     // 底层操作对象
    7.     unsafe = newUnsafe();
    8.     // 创建了channelPipeline
    9.     // 准备跟踪
    10.     pipeline = newChannelPipeline();
    11. }

    DefaultChannelPipeline.java

    1.          // 生成head节点的名称，简单类名后添加#0
    2.     private static final String HEAD_NAME = generateName0(HeadContext.class);
    3.     // 生成tail节点的名称，简单类名后添加#0
    4.     private static final String TAIL_NAME = generateName0(TailContext.class);
    6.         protected DefaultChannelPipeline newChannelPipeline() {
    7.             return new DefaultChannelPipeline(this);
    8.         }
    10.     // pipeline本质上是一个双向链表
    11.     protected DefaultChannelPipeline(Channel channel) {
    12.         this.channel = ObjectUtil.checkNotNull(channel, "channel");
    13.         succeededFuture = new SucceededChannelFuture(channel, null);
    14.         voidPromise =  new VoidChannelPromise(channel, true);
    15.         // 尾节点
    16.           // 准备跟踪
    17.         tail = new TailContext(this);
    18.         // 头节点
    19.         head = new HeadContext(this);
    20.         // 头节的下一个指针指向尾节点
    21.         head.next = tail;
    22.         // 尾节点的上一个指针指向头结点
    23.         tail.prev = head;
    24.     }
    26.  // A special catch-all handler that handles both bytes and messages.
    27.     // tail节点是一个inbound处理器，用于释放资源
    28.     final class TailContext extends AbstractChannelHandlerContext implements ChannelInboundHandler {
    30.         TailContext(DefaultChannelPipeline pipeline) {
    31.             // 第二个参数为当前节点所绑定的eventLoop，当前为null
    32.             // 第三个参数为当前节点的名称
    33.               // 准备跟踪
    34.             super(pipeline, null, TAIL_NAME, TailContext.class);
    35.             // 修改节点状态为ADD_COMPLETE
    36.             setAddComplete();
    37.         }
    38.     }

    AbstractChannelHandContext.java

    1. AbstractChannelHandlerContext(DefaultChannelPipeline pipeline, EventExecutor executor,
    2.                               String name, Class<? extends ChannelHandler> handlerClass) {
    3.     this.name = ObjectUtil.checkNotNull(name, "name");
    4.     this.pipeline = pipeline;
    5.     // 当前节点所绑定的eventLoop
    6.     this.executor = executor;
    7.     // 初始化执行标记，通过该变量可以判断出当前处理器是inbound还是outbound处理器，
    8.     // 可以判断出哪些方法应该跳过，哪些方法应该执行
    9.     // 准备跟踪
    10.     this.executionMask = mask(handlerClass);
    11.     // Its ordered if its driven by the EventLoop or the given Executor is an instanceof OrderedEventExecutor.
    12.     ordered = executor == null || executor instanceof OrderedEventExecutor;
    13. }

    ChannelHeadnlerMask.java

    1. static int mask(Class<? extends ChannelHandler> clazz) {
    2.     // Try to obtain the mask from the cache first. If this fails calculate it and put it in the cache for fast
    3.     // lookup in the future.
    4.     Map<Class<? extends ChannelHandler>, Integer> cache = MASKS.get();
    5.     Integer mask = cache.get(clazz);
    6.     if (mask == null) {
    7.         // 计算mask的值
    8.         mask = mask0(clazz);
    9.         cache.put(clazz, mask);
    10.     }
    11.     return mask;
    12. }
    13.  /**
    14.      * Calculate the {@code executionMask}.
    15.      */
    16.     private static int mask0(Class<? extends ChannelHandler> handlerType) {
    17.         // 初始化mask
    18.         int mask = MASK_EXCEPTION_CAUGHT;
    19.         try {
    20.             // 处理handlerType为inbound处理器的情况
    21.             if (ChannelInboundHandler.class.isAssignableFrom(handlerType)) {
    22.                 mask |= MASK_ALL_INBOUND;
    24.                 // 判断channelRegistered()方法是否应该跳过执行，即不执行
    25.                 // 若channelRegistered()上出现了@Skip注解，则应该跳过
    26.                 // 像我们自定义的Handler处理器里面继承的ChannelInboundHandlerAdapter
    27.                 // 类里面的channelRead(),exceptionCaught等等它就不会跳过，
    28.                 // 因为我们实现的方法上面没有@Skip注解
    29.                 if (isSkippable(handlerType, "channelRegistered", ChannelHandlerContext.class)) {
    30.                     mask &= ~MASK_CHANNEL_REGISTERED;
    31.                 }
    32.                 if (isSkippable(handlerType, "channelUnregistered", ChannelHandlerContext.class)) {
    33.                     mask &= ~MASK_CHANNEL_UNREGISTERED;
    34.                 }
    35.                 if (isSkippable(handlerType, "channelActive", ChannelHandlerContext.class)) {
    36.                     mask &= ~MASK_CHANNEL_ACTIVE;
    37.                 }
    38.                 if (isSkippable(handlerType, "channelInactive", ChannelHandlerContext.class)) {
    39.                     mask &= ~MASK_CHANNEL_INACTIVE;
    40.                 }
    41.                 if (isSkippable(handlerType, "channelRead", ChannelHandlerContext.class, Object.class)) {
    42.                     mask &= ~MASK_CHANNEL_READ;
    43.                 }
    44.                 if (isSkippable(handlerType, "channelReadComplete", ChannelHandlerContext.class)) {
    45.                     mask &= ~MASK_CHANNEL_READ_COMPLETE;
    46.                 }
    47.                 if (isSkippable(handlerType, "channelWritabilityChanged", ChannelHandlerContext.class)) {
    48.                     mask &= ~MASK_CHANNEL_WRITABILITY_CHANGED;
    49.                 }
    50.                 if (isSkippable(handlerType, "userEventTriggered", ChannelHandlerContext.class, Object.class)) {
    51.                     mask &= ~MASK_USER_EVENT_TRIGGERED;
    52.                 }
    53.             }
    55.             // 处理handlerType为outbound处理器的情况
    56.             if (ChannelOutboundHandler.class.isAssignableFrom(handlerType)) {
    57.                 mask |= MASK_ALL_OUTBOUND;
    59.                 if (isSkippable(handlerType, "bind", ChannelHandlerContext.class,
    60.                         SocketAddress.class, ChannelPromise.class)) {
    61.                     mask &= ~MASK_BIND;
    62.                 }
    63.                 if (isSkippable(handlerType, "connect", ChannelHandlerContext.class, SocketAddress.class,
    64.                         SocketAddress.class, ChannelPromise.class)) {
    65.                     mask &= ~MASK_CONNECT;
    66.                 }
    67.                 if (isSkippable(handlerType, "disconnect", ChannelHandlerContext.class, ChannelPromise.class)) {
    68.                     mask &= ~MASK_DISCONNECT;
    69.                 }
    70.                 if (isSkippable(handlerType, "close", ChannelHandlerContext.class, ChannelPromise.class)) {
    71.                     mask &= ~MASK_CLOSE;
    72.                 }
    73.                 if (isSkippable(handlerType, "deregister", ChannelHandlerContext.class, ChannelPromise.class)) {
    74.                     mask &= ~MASK_DEREGISTER;
    75.                 }
    76.                 if (isSkippable(handlerType, "read", ChannelHandlerContext.class)) {
    77.                     mask &= ~MASK_READ;
    78.                 }
    79.                 if (isSkippable(handlerType, "write", ChannelHandlerContext.class,
    80.                         Object.class, ChannelPromise.class)) {
    81.                     mask &= ~MASK_WRITE;
    82.                 }
    83.                 if (isSkippable(handlerType, "flush", ChannelHandlerContext.class)) {
    84.                     mask &= ~MASK_FLUSH;
    85.                 }
    86.             }
    88.             if (isSkippable(handlerType, "exceptionCaught", ChannelHandlerContext.class, Throwable.class)) {
    89.                 mask &= ~MASK_EXCEPTION_CAUGHT;
    90.             }
    91.         } catch (Exception e) {
    92.             // Should never reach here.
    93.             PlatformDependent.throwException(e);
    94.         }
    96.         return mask;
    97.     }
    99.     // 判断方法上是否有@Skip注解
    100.     @SuppressWarnings("rawtypes")
    101.     private static boolean isSkippable(
    102.             final Class<?> handlerType, final String methodName, final Class<?>... paramTypes) throws Exception {
    103.         return AccessController.doPrivileged(new PrivilegedExceptionAction<Boolean>() {
    104.             @Override
    105.             public Boolean run() throws Exception {
    106.                 return handlerType.getMethod(methodName, paramTypes).isAnnotationPresent(Skip.class);
    107.             }
    108.         });
    109.     }

    回过头来跟踪头节点

    DefaultChannelPipeline.java

    1. // head节点既是一个inbound处理器，又是一个outbound处理器
    2. // 作为inbound处理器，其用于将消息向下一个节点传递
    3. // 作为outbound处理器，其用于完成直接的底层操作
    4. final class HeadContext extends AbstractChannelHandlerContext
    5.         implements ChannelOutboundHandler, ChannelInboundHandler {
    7.     private final Unsafe unsafe;
    9.     HeadContext(DefaultChannelPipeline pipeline) {
    10.         super(pipeline, null, HEAD_NAME, HeadContext.class);
    11.         // 获取底层操作对象unsafe
    12.         unsafe = pipeline.channel().unsafe();
    13.         setAddComplete();
    14.     }
    15. }

    像Pipeline中添加处理器

    DefaultChannelPipeline.java

    1. @Override
    2. public final ChannelPipeline addLast(ChannelHandler... handlers) {
    3.     // 第一个参数是一个EventLoopGroup
    4.     // 第二个参数是一个数组
    5.     return addLast(null, handlers);
    6. }
    8. @Override
    9. public final ChannelPipeline addLast(EventExecutorGroup executor, ChannelHandler... handlers) {
    10.     if (handlers == null) {
    11.         throw new NullPointerException("handlers");
    12.     }
    14.     // 逐个添加处理器
    15.     for (ChannelHandler h: handlers) {
    16.         if (h == null) {
    17.             break;
    18.         }
    19.         // 添加一个处理器到pipeline
    20.         // 第二个参数为name，
    21.         addLast(executor, null, h);
    22.     }
    24.     return this;
    25. }
    27.  @Override
    28.     public final ChannelPipeline addLast(EventExecutorGroup group, String name, ChannelHandler handler) {
    29.         final AbstractChannelHandlerContext newCtx;
    30.         synchronized (this) {
    31.             // 检测当前handler是否被重复添加
    32.             // 1:准备跟踪
    33.             checkMultiplicity(handler);
    34.             // 将处理器封装为一个节点
    35.             // filterName() 获取节点名称
    36.               // 2:准备跟踪
    37.             newCtx = newContext(group, filterName(name, handler), handler);
    38.             // 将新的节点添加到pipeline
    39.             addLast0(newCtx);
    41.             // If the registered is false it means that the channel was not registered on an eventLoop yet.
    42.             // In this case we add the context to the pipeline and add a task that will call
    43.             // ChannelHandler.handlerAdded(...) once the channel is registered.
    44.             // 处理channel没有注册的情况
    45.             if (!registered) {
    46.                 newCtx.setAddPending();
    47.                 callHandlerCallbackLater(newCtx, true);
    48.                 return this;
    49.             }
    50.             // 获取当前处理器节点的eventLoop
    51.             EventExecutor executor = newCtx.executor();
    52.             if (!executor.inEventLoop()) {
    53.                 callHandlerAddedInEventLoop(newCtx, executor);
    54.                 return this;
    55.             }
    56.         }
    57.         // 触发当前处理器的handlerAdded()方法
    58.           // 3:准备跟踪
    59.         callHandlerAdded0(newCtx);
    60.         return this;
    61.     }
    63.     private static void checkMultiplicity(ChannelHandler handler) {
    64.         if (handler instanceof ChannelHandlerAdapter) {
    65.             ChannelHandlerAdapter h = (ChannelHandlerAdapter) handler;
    66.             // 若当前处理器是非共享处理器，且已经被添加过了，则抛出异常
    67.             // 因为非共享处理器只能被添加一次
    68.             if (!h.isSharable() && h.added) {
    69.                 throw new ChannelPipelineException(
    70.                         h.getClass().getName() +
    71.                         " is not a @Sharable handler, so can't be added or removed multiple times.");
    72.             }
    73.             // 修改添加状态变量
    74.             h.added = true;
    75.         }
    76.     }
    79.     private AbstractChannelHandlerContext newContext(EventExecutorGroup group, String name, ChannelHandler handler) {
    80.         // childExecutor() 获取当前节点所绑定的eventLoop
    81.         return new DefaultChannelHandlerContext(this, childExecutor(group), name, handler);
    82.     }
    85. private void callHandlerAdded0(final AbstractChannelHandlerContext ctx) {
    86.         try {
    87.             // 调用当前节点处理器的handlerAdded()
    88.             ctx.callHandlerAdded();
    89.         } catch (Throwable t) {
    90.             boolean removed = false;
    91.             try {
    92.                 remove0(ctx);
    93.                 ctx.callHandlerRemoved();
    94.                 removed = true;
    95.             } catch (Throwable t2) {
    96.                 if (logger.isWarnEnabled()) {
    97.                     logger.warn("Failed to remove a handler: " + ctx.name(), t2);
    98.                 }
    99.             }
    101.             if (removed) {
    102.                 fireExceptionCaught(new ChannelPipelineException(
    103.                         ctx.handler().getClass().getName() +
    104.                         ".handlerAdded() has thrown an exception; removed.", t));
    105.             } else {
    106.                 fireExceptionCaught(new ChannelPipelineException(
    107.                         ctx.handler().getClass().getName() +
    108.                         ".handlerAdded() has thrown an exception; also failed to remove.", t));
    109.             }
    110.         }
    111.     }

    处理器在Pipeline中被删除

    ChannelInitializer.java

    1. // 其为一个可共享的处理器，即该处理器实例可以被添加到多个pipeline，可以被多次添加到pipeline
    2. @Sharable
    3. // 其为一个inbound处理器
    4. public abstract class ChannelInitializer<C extends Channel> extends ChannelInboundHandlerAdapter {
    6.     private static final InternalLogger logger = InternalLoggerFactory.getInstance(ChannelInitializer.class);
    7.     // We use a Set as a ChannelInitializer is usually shared between all Channels in a Bootstrap /
    8.     // ServerBootstrap. This way we can reduce the memory usage compared to use Attributes.
    9.     // 该集合中存放的是当前处理器实例所封装的处理器节点
    10.     private final Set<ChannelHandlerContext> initMap = Collections.newSetFromMap(
    11.             new ConcurrentHashMap<ChannelHandlerContext, Boolean>());
    13.     // 当, 当前处理器被添加到pipeline后，该方法就会被调用
    14.     @Override
    15.     public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
    16.         if (ctx.channel().isRegistered()) {  // 若当前channel已经注册完毕了，则...
    17.             if (initChannel(ctx)) {  // 调用本地的initChannel()
    18.                 removeState(ctx);  // 将当前处理器所封装的节点从initMap中删除
    19.             }
    20.         }
    21.     }
    23.  /**
    24.      * @param ctx  当前处理器所封装为的节点
    25.      * @return
    26.      * @throws Exception
    27.      */
    28.     @SuppressWarnings("unchecked")
    29.     private boolean initChannel(ChannelHandlerContext ctx) throws Exception {
    30.         // 若ctx节点添加到initMap成功，则...
    31.         if (initMap.add(ctx)) { // Guard against re-entrance.
    32.             try {
    33.                 // 调用重写的initChannel()，该方法执行完毕，则其历史使命完成，就可以做收尾工作了
    34.                 initChannel((C) ctx.channel());
    35.             } catch (Throwable cause) {
    36.                 // Explicitly call exceptionCaught(...) as we removed the handler before calling initChannel(...).
    37.                 // We do so to prevent multiple calls to initChannel(...).
    38.                 exceptionCaught(ctx, cause);
    39.             } finally {
    40.                 // 收尾工作：将当前节点从当前pipeline上删除
    41.                 ChannelPipeline pipeline = ctx.pipeline();
    42.                 // 从pipeline中首先对当前处理器进行查询，若存在，则将该节点从pipeline中删除
    43.                 if (pipeline.context(this) != null) {
    44.                       // 准备跟踪
    45.                     pipeline.remove(this);
    46.                 }
    47.             }
    48.             return true;
    49.         }
    50.         return false;
    51.     }
    52. }

    DefaultChannelPipeline.java

    1. @Override
    2. public final ChannelPipeline remove(ChannelHandler handler) {
    3.       // 跟踪getContextOrDie()方法
    4.     remove(getContextOrDie(handler));
    5.     return this;
    6. }
    8.     // 不成功便成仁
    9.     private AbstractChannelHandlerContext getContextOrDie(ChannelHandler handler) {
    10.         // 查询当前处理器所封装的节点
    11.           // 准备跟踪context()方法
    12.         AbstractChannelHandlerContext ctx = (AbstractChannelHandlerContext) context(handler);
    13.         // 若没有找到，则直接抛出异常，否则返回找到的节点
    14.         if (ctx == null) {
    15.             throw new NoSuchElementException(handler.getClass().getName());
    16.         } else {
    17.             return ctx;
    18.         }
    19.     }
    22.  @Override
    23.     public final ChannelHandlerContext context(ChannelHandler handler) {
    24.         if (handler == null) {
    25.             throw new NullPointerException("handler");
    26.         }
    28.         // 从头节点的下一个节点开始
    29.         AbstractChannelHandlerContext ctx = head.next;
    30.         // 遍历整个pipeline，要么找到了返回，要么返回null
    31.         for (;;) {
    33.             if (ctx == null) {
    34.                 return null;
    35.             }
    37.             if (ctx.handler() == handler) {
    38.                 return ctx;
    39.             }
    40.             // 获取下一个节点
    41.             ctx = ctx.next;
    42.         }
    43.     }

    然后回过头来再跟踪一下ChannelInitializer.java中的removeState()

    1. private void removeState(final ChannelHandlerContext ctx) {
    2.     // The removal may happen in an async fashion if the EventExecutor we use does something funky.
    3.     // 若当前节点的状态为 删除完毕，则将当前节点从initMap中删除
    4.     if (ctx.isRemoved()) {
    5.         initMap.remove(ctx);
    6.     } else {
    7.         // The context is not removed yet which is most likely the case because a custom EventExecutor is used.
    8.         // Let's schedule it on the EventExecutor to give it some more time to be completed in case it is offloaded.
    9.         ctx.executor().execute(new Runnable() {
    10.             @Override
    11.             public void run() {
    12.                 initMap.remove(ctx);
    13.             }
    14.         });
    15.     }
    16. }