NETTY组件-CHANNEL

/*绑定到端口，阻塞等待直到连接完成*/
ChannelFuture f = b.bind().sync();

serverBootstrap.bind(port).addListener(future -> {
            if (future.isSuccess()) {
                System.out.println(new Date() + ": 端口[" + port + "]绑定成功!");
            } else {
                System.err.println("端口[" + port + "]绑定失败!");
            }

Channel channel = ...
    ChannelFuture future = channel.connect(new InetSocketAddress("192.168.0.1", 6666));
// 注册一个监听器
future.addListener(new ChannelFutureListener() {
    @Override
    public void operationComplete(ChannelFuture future) {
        if (future.isSuccess()) {
            // do something....
        } else {
            // 输出错误信息
            Throwable cause = future.cause();
            cause.printStackTrace();
            // do something....
        }
    }
});

public interface ChannelFutureListener extends GenericFutureListener<ChannelFuture> {
    // 在Future完成时关闭
    ChannelFutureListener CLOSE = new ChannelFutureListener() {
        public void operationComplete(ChannelFuture future) {
            future.channel().close();
        }
    };
    // 如果失败则关闭
    ChannelFutureListener CLOSE_ON_FAILURE = new ChannelFutureListener() {
        public void operationComplete(ChannelFuture future) {
            if(!future.isSuccess()) {
                future.channel().close();
            }

        }
    };
    // 将异常信息传递给下一个ChannelHandler
    ChannelFutureListener FIRE_EXCEPTION_ON_FAILURE = new ChannelFutureListener() {
        public void operationComplete(ChannelFuture future) {
            if(!future.isSuccess()) {
                future.channel().pipeline().fireExceptionCaught(future.cause());
            }

        }
    };
}

public interface ChannelHandler {
    void handlerAdded(ChannelHandlerContext var1) throws Exception;

    void handlerRemoved(ChannelHandlerContext var1) throws Exception;

    /** @deprecated */
    @Deprecated
    void exceptionCaught(ChannelHandlerContext var1, Throwable var2) throws Exception;

    // 该注解表明这个ChannelHandler可被其他线程复用
    @Inherited
    @Documented
    @Target({ElementType.TYPE})
    @Retention(RetentionPolicy.RUNTIME)
    public @interface Sharable {
    }
}

public interface ChannelInboundHandler extends ChannelHandler {
    // 当channel被注册到EventLoop时被调用
    void channelRegistered(ChannelHandlerContext var1) throws Exception;

    // 当channel已经被创建，但还未注册到EventLoop（或者从EventLoop中注销）被调用
    void channelUnregistered(ChannelHandlerContext var1) throws Exception;

    // 当channel处于活动状态（连接到远程节点）被调用
    void channelActive(ChannelHandlerContext var1) throws Exception;

    // 当channel处于非活动状态（没有连接到远程节点）被调用
    void channelInactive(ChannelHandlerContext var1) throws Exception;

    // 当从channel读取数据时被调用
    void channelRead(ChannelHandlerContext var1, Object var2) throws Exception;

    // 当channel的上一个读操作完成时被调用
    void channelReadComplete(ChannelHandlerContext var1) throws Exception;

    // 当ChannelInboundHandler.fireUserEventTriggered()方法被调用时被调用
    void userEventTriggered(ChannelHandlerContext var1, Object var2) throws Exception;

    // 当channel的可写状态发生改变时被调用
    void channelWritabilityChanged(ChannelHandlerContext var1) throws Exception;

    // 当处理过程中发生异常时被调用
    void exceptionCaught(ChannelHandlerContext var1, Throwable var2) throws Exception;
}
 
package io.netty.channel;

        import io.netty.channel.ChannelHandler;
        import io.netty.channel.ChannelHandlerContext;
        import io.netty.channel.ChannelPromise;
        import java.net.SocketAddress;

public interface ChannelOutboundHandler extends ChannelHandler {

    // 当请求将Channel绑定到一个地址时被调用
// ChannelPromise是ChannelFuture的一个子接口，定义了如setSuccess(),setFailure()等方法
    void bind(ChannelHandlerContext var1, SocketAddress var2, ChannelPromise var3) throws Exception;

    // 当请求将Channel连接到远程节点时被调用
    void connect(ChannelHandlerContext var1, SocketAddress var2, SocketAddress var3, ChannelPromise var4) throws Exception;

    // 当请求将Channel从远程节点断开时被调用
    void disconnect(ChannelHandlerContext var1, ChannelPromise var2) throws Exception;

    // 当请求关闭Channel时被调用
    void close(ChannelHandlerContext var1, ChannelPromise var2) throws Exception;

    // 当请求将Channel从它的EventLoop中注销时被调用
    void deregister(ChannelHandlerContext var1, ChannelPromise var2) throws Exception;

    // 当请求从Channel读取数据时被调用
    void read(ChannelHandlerContext var1) throws Exception;

    // 当请求通过Channel将数据写到远程节点时被调用
    void write(ChannelHandlerContext var1, Object var2, ChannelPromise var3) throws Exception;

    // 当请求通过Channel将缓冲中的数据冲刷到远程节点时被调用
    void flush(ChannelHandlerContext var1) throws Exception;
}

public ChannelHandlerAdapter(){
        }

// 该方法不允许将此ChannelHandler共享复用
protected void ensureNotSharable(){
        if(this.isSharable()){
        throw new IllegalStateException("ChannelHandler "+this.getClass().getName()+" is not allowed to be shared");
        }
        }

// 使用反射判断实现类有没有@Sharable注解，以确认该类是否为可共享复用的
public boolean isSharable(){
        Class clazz=this.getClass();
        Map cache=InternalThreadLocalMap.get().handlerSharableCache();
        Boolean sharable=(Boolean)cache.get(clazz);
        if(sharable==null){
        sharable=Boolean.valueOf(clazz.isAnnotationPresent(Sharable.class));
        cache.put(clazz,sharable);
        }

        return sharable.booleanValue();
        }

public void handlerAdded(ChannelHandlerContext ctx)throws Exception{
        }

public void handlerRemoved(ChannelHandlerContext ctx)throws Exception{
        }

public void exceptionCaught(ChannelHandlerContext ctx,Throwable cause)throws Exception{
        ctx.fireExceptionCaught(cause);
        }
        }

public class ChannelInboundHandlerAdapter extends ChannelHandlerAdapter implements ChannelInboundHandler {
    public ChannelInboundHandlerAdapter() {
    }

    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        ctx.fireChannelRegistered();
    }

    public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
        ctx.fireChannelUnregistered();
    }

    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        ctx.fireChannelActive();
    }

    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        ctx.fireChannelInactive();
    }

    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ctx.fireChannelRead(msg);
    }

    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.fireChannelReadComplete();
    }

    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        ctx.fireUserEventTriggered(evt);
    }

    public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
        ctx.fireChannelWritabilityChanged();
    }

    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.fireExceptionCaught(cause);
    }
}

public class ChannelOutboundHandlerAdapter extends ChannelHandlerAdapter implements ChannelOutboundHandler {
    public ChannelOutboundHandlerAdapter() {
    }

    public void bind(ChannelHandlerContext ctx, SocketAddress localAddress, ChannelPromise promise) throws Exception {
        ctx.bind(localAddress, promise);
    }

    public void connect(ChannelHandlerContext ctx, SocketAddress remoteAddress, SocketAddress localAddress, ChannelPromise promise) throws Exception {
        ctx.connect(remoteAddress, localAddress, promise);
    }

    public void disconnect(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
        ctx.disconnect(promise);
    }

    public void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
        ctx.close(promise);
    }

    public void deregister(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
        ctx.deregister(promise);
    }

    public void read(ChannelHandlerContext ctx) throws Exception {
        ctx.read();
    }

    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
        ctx.write(msg, promise);
    }

    public void flush(ChannelHandlerContext ctx) throws Exception {
        ctx.flush();
    }
}

// 这个泛型为消息对象的类型
public abstract class SimpleChannelInboundHandler<I> extends ChannelInboundHandlerAdapter {
    private final TypeParameterMatcher matcher;
    private final boolean autoRelease;

    protected SimpleChannelInboundHandler() {
        this(true);
    }

    protected SimpleChannelInboundHandler(boolean autoRelease) {
        this.matcher = TypeParameterMatcher.find(this, SimpleChannelInboundHandler.class, "I");
        this.autoRelease = autoRelease;
    }

    protected SimpleChannelInboundHandler(Class<? extends I> inboundMessageType) {
        this(inboundMessageType, true);
    }

    protected SimpleChannelInboundHandler(Class<? extends I> inboundMessageType, boolean autoRelease) {
        this.matcher = TypeParameterMatcher.get(inboundMessageType);
        this.autoRelease = autoRelease;
    }

    public boolean acceptInboundMessage(Object msg) throws Exception {
        return this.matcher.match(msg);
    }

    // SimpleChannelInboundHandler只是替你做了ReferenceCountUtil.release()
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        boolean release = true;

        try {
            if (this.acceptInboundMessage(msg)) {
                this.channelRead0(ctx, msg);
            } else {
                release = false;
                ctx.fireChannelRead(msg);
            }
        } finally {
            if (this.autoRelease && release) {
                ReferenceCountUtil.release(msg);
            }

        }

    }

    // 这个方法才是我们需要实现的方法
    protected abstract void channelRead0(ChannelHandlerContext var1, I var2) throws Exception;
}

    // ReferenceCountUtil中的源码，release方法对消息对象的类型进行判断然后调用它的release()方法
    public static boolean release(Object msg) {
        return msg instanceof ReferenceCounted ? ((ReferenceCounted) msg).release() : false;
    }

public class DefaultChannelPipeline implements ChannelPipeline {
.........
    // 头部节点和尾部节点的引用变量
// ChannelHandlerContext在ChannelPipeline中是以链表的形式组织的
    final AbstractChannelHandlerContext head;
    final AbstractChannelHandlerContext tail;
.........

    // 添加一个ChannelHandler到链表尾部
    public final ChannelPipeline addLast(String name, ChannelHandler handler) {
        return this.addLast((EventExecutorGroup) null, name, handler);
    }

    public final ChannelPipeline addLast(EventExecutorGroup group, String name, ChannelHandler handler) {
        final AbstractChannelHandlerContext newCtx;
        synchronized (this) {
// 检查ChannelHandler是否为一个共享对象(@Sharable)
// 如果该ChannelHandler没有@Sharable注解，并且是已被添加过的那么就抛出异常
            checkMultiplicity(handler);
// 返回一个DefaultChannelHandlerContext，注意该对象持有了传入的ChannelHandler
            newCtx = this.newContext(group, this.filterName(name, handler), handler);
            this.addLast0(newCtx);
// 如果当前ChannelPipeline没有被注册，那么就先加到未决链表中
            if (!this.registered) {
                newCtx.setAddPending();
                this.callHandlerCallbackLater(newCtx, true);
                return this;
            }

// 否则就调用ChannelHandler中的handlerAdded()
            EventExecutor executor = newCtx.executor();
            if (!executor.inEventLoop()) {
                newCtx.setAddPending();
                executor.execute(new Runnable() {
                    public void run() {
                        DefaultChannelPipeline.this.callHandlerAdded0(newCtx);
                    }
                });
                return this;
            }
        }

        this.callHandlerAdded0(newCtx);
        return this;
    }

    // 将新的ChannelHandlerContext插入到尾部与尾部之前的节点之间
    private void addLast0(AbstractChannelHandlerContext newCtx) {
        AbstractChannelHandlerContext prev = this.tail.prev;
        newCtx.prev = prev;
        newCtx.next = this.tail;
        prev.next = newCtx;
        this.tail.prev = newCtx;
    }
.....
}

public class DefaultChannelPipeline implements ChannelPipeline {

    public final ChannelPipeline fireChannelRead(Object msg) {
// 注意这里将头节点传入了进去
        AbstractChannelHandlerContext.invokeChannelRead(this.head, msg);
        return this;
    }

}

abstract class AbstractChannelHandlerContext extends DefaultAttributeMap implements ChannelHandlerContext, ResourceLeakHint {

    static void invokeChannelRead(final AbstractChannelHandlerContext next, Object msg) {
        final Object m = next.pipeline.touch(ObjectUtil.checkNotNull(msg, "msg"), next);
        EventExecutor executor = next.executor();
        if (executor.inEventLoop()) {
            next.invokeChannelRead(m);
        } else {
            executor.execute(new Runnable() {
                public void run() {
                    next.invokeChannelRead(m);
                }
            });
        }

    }

    private void invokeChannelRead(Object msg) {
        if (this.invokeHandler()) {
            try {
                ((ChannelInboundHandler) this.handler()).channelRead(this, msg);
            } catch (Throwable var3) {
                this.notifyHandlerException(var3);
            }
        } else {
// 寻找下一个ChannelHandler
            this.fireChannelRead(msg);
        }

    }

    public ChannelHandlerContext fireChannelRead(Object msg) {
        invokeChannelRead(this.findContextInbound(), msg);
        return this;
    }

    private AbstractChannelHandlerContext findContextInbound() {
        AbstractChannelHandlerContext ctx = this;

        do {
            ctx = ctx.next;
        } while (!ctx.inbound); // 直到找到一个ChannelInboundHandler

        return ctx;
    }

}