netty入门

1 阻塞IO ，阻塞异步IO，非阻塞异步IO

首先学习netty ,那就必须对阻塞IO ，阻塞异步IO，非阻塞异步IO有一个了解，可以参考https://www.ibm.com/developerworks/cn/linux/l-async/

2 netty的架构

netty吸收了多种协议( FTP,SMTP,HTTP,WebSocket,SPDY,二进制、文本)实现的经验，帮助实现了构建快速构建一个高性能、稳定、以扩展的，基于NIO的客户端、服务端框架，其体系结构如下：

 3 核心类分析

1) EventLoop类

在netty中的体系架构中的位置

 EventLoop是基于current包中的ScheduledExecutorService来做线程调度的:下面是EventLoop的源码分析：

EventLoop

 

public interface EventLoop extends EventExecutor, EventLoopGroup {
    @Override
    EventLoopGroup parent();
}

 eventLoop继承EventLoopGroup，EventLoopGroup的源码如下：

 

EventLoopGroup

 

public interface EventLoopGroup extends EventExecutorGroup {
    /**
     * 下一个事件执行器(EventExcutor)
     */
    @Override
    EventLoop next();

    /**
     * 将一个Channel注册给EventExecutor,返回一个ChannelFuture
     */
    ChannelFuture register(Channel channel);

    /**
     * Channel注册给EventExecutor,ChannelPromise是一个可以写入的ChannelFuture* 返回一个ChannelFuture
     */
    ChannelFuture register(Channel channel, ChannelPromise promise);
}

EventExecutor

public interface EventExecutor extends EventExecutorGroup {

    /**
     * 返回指向自己的引用
     */
    @Override
    EventExecutor next();

    /**
     * 返回父类EventExecutorGroup
     */
    EventExecutorGroup parent();

    /**
     * 判断一个线程是否在ScheduledExecutorService的调度中
     */
    boolean inEventLoop();

    /**
     * 判断一个线程是否在ScheduleExecutorService的调度中
     */
    boolean inEventLoop(Thread thread);
    <V> Promise<V> newPromise();
    <V> ProgressivePromise<V> newProgressivePromise();
    <V> Future<V> newSucceededFuture(V result);
    <V> Future<V> newFailedFuture(Throwable cause);
}

 

 

EventLoop下就有EmbededEventLoop和抽象类SingleThreadEventLoop 
SingleThreadEventLoop

根据字面意思，可以看出管理event的线程是单一的。他的子类有AioEventLoop，LocalEventLoop，NioEventLoop，OioEventLoopSingleThreadEventLoopImp等。 

 

MultithreadEventLoopGroupl

实际上就是SingleThreadEventLoop

详情可以参考:

http://airu.iteye.com/blog/1749331

2) ChannelHandler类

 

public interface ChannelHandler {

    /**
     * Gets called after the {@link ChannelHandler} was added to the actual context and it's ready to handle events.
     */
    void handlerAdded(ChannelHandlerContext ctx) throws Exception;

    /**
     * Gets called after the {@link ChannelHandler} was removed from the actual context and it doesn't handle events
     * anymore.
     */
    void handlerRemoved(ChannelHandlerContext ctx) throws Exception;

    /**
     * Gets called if a {@link Throwable} was thrown.
     *
     * @deprecated  Will be removed in the future and only {@link ChannelInboundHandler} will receive
     *              exceptionCaught events. For {@link ChannelOutboundHandler} the {@link ChannelPromise}
     *              must be failed.
     */
    @Deprecated
    void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception;

    /**
     * Indicates that the same instance of the annotated {@link ChannelHandler}
     * can be added to one or more {@link ChannelPipeline}s multiple times
     * without a race condition.
     * <p>
     * If this annotation is not specified, you have to create a new handler
     * instance every time you add it to a pipeline because it has unshared
     * state such as member variables.
     * <p>
     * This annotation is provided for documentation purpose, just like
     * <a href="http://www.javaconcurrencyinpractice.com/annotations/doc/">the JCIP annotations</a>.
     */
    @Inherited
    @Documented
    @Target(ElementType.TYPE)
    @Retention(RetentionPolicy.RUNTIME)
    @interface Sharable {
        // no value
    }
}

每一个ChannelHandler添加都ChannelPipeline中都需要创建和绑定一个ChannelHandlerContext,如图所示:

 

 ChannelHandlerContext类

 

public interface ChannelHandlerContext
         extends AttributeMap, ChannelPropertyAccess,
                 ChannelInboundInvoker, ChannelOutboundInvoker {

    /**
     * Return the {@link Channel} which is bound to the {@link ChannelHandlerContext}.
     */
    Channel channel();

    /**
     * The {@link EventExecutor} that is used to dispatch the events. This can also be used to directly
     * submit tasks that get executed in the event loop. For more information please refer to the
     * {@link EventExecutor} javadoc.
     */
    EventExecutor executor();

    /**
     * The unique name of the {@link ChannelHandlerContext}.The name was used when then {@link ChannelHandler}
     * was added to the {@link ChannelPipeline}. This name can also be used to access the registered
     * {@link ChannelHandler} from the {@link ChannelPipeline}.
     */
    String name();

    /**
     * The {@link ChannelHandler} that is bound this {@link ChannelHandlerContext}.
     */
    ChannelHandler handler();

    /**
     * Return {@code true} if the {@link ChannelHandler} which belongs to this {@link ChannelHandler} was removed
     * from the {@link ChannelPipeline}. Note that this method is only meant to be called from with in the
     * {@link EventLoop}.
     */
    boolean isRemoved();

    @Override
    ChannelHandlerContext fireChannelRegistered();

    @Override
    @Deprecated
    ChannelHandlerContext fireChannelUnregistered();

    @Override
    ChannelHandlerContext fireChannelActive();

    @Override
    ChannelHandlerContext fireChannelInactive();

    @Override
    ChannelHandlerContext fireExceptionCaught(Throwable cause);

    @Override
    ChannelHandlerContext fireUserEventTriggered(Object event);

    @Override
    ChannelHandlerContext fireChannelRead(Object msg);

    @Override
    ChannelHandlerContext fireChannelReadComplete();

    @Override
    ChannelHandlerContext fireChannelWritabilityChanged();

    @Override
    ChannelHandlerContext flush();
}

 

 

这里的ChannelInboundInvoker接口基本上就是拥有通知Channel上的各种事件的能力，二ChannelOutboundInvoker则是具备了向管道外写数据，连接等功能。注意返回值都是ChannelFuture这就说明这些操作都是异步的。 
总之ChannelOutboundInvoker是Channel，ChannelhandlerContext, ChannelPipeline的父接口。 
对于ChannelHandlerContext来说，基本上就是对传输过程中的各个方面的管理器。包括ChannelHandler的注册销毁等等。 

每个ChannelHander都是有方向性的。这里的方向性，也就是管道中inbound和outbound的意思。所以ChannelHander的两个子接口 ChannelOperationHandler 也就是outbound 和ChannelStateHandler 也就是inbound。围绕着in与out，又催生出子接口 ChannelOutboundHandler 和 ChannelInboundHandler。比较常用的子类:

ChannelInboundHandlerAdapter 和 ChannelOutboundHandlerAdapter

 

public class ChannelInboundHandlerAdapter extends ChannelHandlerAdapter implements ChannelInboundHandler {

    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);
    }
}

 可以根据自己需要重写对应的方法,(方法的使用，请查阅api),ChannelOutboundHandlerAdapter也类似

ChannelPipeline类

ChannelPipeline的作用看如下图就可以明白了

 Buffer类

netty重写了NIO中的buffer，NIO中的buffer使用起来容易出错，也比较复杂，垃圾回收起来不方便，netty 中的Buffer则解决了这些问题，内存分配算法可以参见：http://en.wikipedia.org/wiki/Buddy_memory_allocation

Bootstrap类

Bootstrap 和 BootstrapServer都是继承AbstractBootstrap

查看AbstractBootstrap的变量定义

    private volatile EventLoopGroup group;
    private volatile ChannelFactory<? extends C> channelFactory;
    private volatile SocketAddress localAddress;
    private final Map<ChannelOption<?>, Object> options = new LinkedHashMap<ChannelOption<?>, Object>();
    private final Map<AttributeKey<?>, Object> attrs = new LinkedHashMap<AttributeKey<?>, Object>();
    private volatile ChannelHandler handler;

 一个事件组，netty是事件驱动的，事件组应该是各种事件的容器；一个Channel工厂，这自然是用来生产Channel的（这里应该有各种管道。比如服务端的和客户端的就不一样） ；SocketAddress是本地地址 options 是这个管道的属性集合 attrs 是属性键集合（这个搞不懂，先放下）； ChannelHandler 用于处理管道的各种事件吧。 

具体方法的含义如下：

1, public B group(EventLoopGroup group) 

这个函数简单，设置group，注意如果传入参数为空，或者原先的group不为空都要抛出异常。然后，返回自身 ，注意泛型 B 

2, public B channel(Class<? extends Channel> channelClass) 

这个类用来创建一个Channel实例，使用了ChannelFactory。使用工厂模式推迟具体实例创建到子类中。这里可以看做一个工厂设置 

3, public B channelFactory(ChannelFactory factory) 

2调用了3， 

4，public B localAddress(SocketAddress localAddress) 

绑定本地地址 

5， public B localAddress(int port) 

同样绑定本地地址 

6， public B localAddress(String host, int port) 

同4,5 

7， public  B localAddress(InetAddress host, int port) 

通4,5,6 

8， public <T> B option(ChannelOption<T> option, T value) 

实例化Channel时的参数。如果 value为null，这个option被remove掉。 

9， public <T> B attr(Attribute<T> key, T value) 

实例化Channel后Channel的属性value为null时，attr被remove掉。 

10， public void shutdown() 

这里关闭的是EventLoopGroup 

11， protected void validate() 

这里变protected了，用于同类的参数验证。子类如果复写，需要使用super先调用父类 

12，protected final void validate(ChannelFuture future) 

验证了future不为null，然后调用11， 不可以被复写 

13，public ChannelFuture bind() 

创建一个Channel并且绑定 

14，public B handler(ChannelHandler handler) 

handler不能为null，设置handler 

15，public static boolean ensureOpen(ChannelFuture future) 

确保future中的channel为open状态，否则futrue设置为失败ChannelException 

16，public abstract ChannelFuture bind(ChannelFuture future) 

把Channel绑定到给定的ChannelFactory上。抽象方法，意味着需要具体的Channel 

Bootstrap中多了两个变量

    private static final InternalLogger logger = InternalLoggerFactory.getInstance(Bootstrap.class);

    private volatile SocketAddress remoteAddress;

 BootstrapServer类

    private final Map<ChannelOption<?>, Object> childOptions = new LinkedHashMap<ChannelOption<?>, Object>();
    private final Map<AttributeKey<?>, Object> childAttrs = new LinkedHashMap<AttributeKey<?>, Object>();
    private volatile EventLoopGroup childGroup;
    private volatile ChannelHandler childHandler;

 

    /**
     * Set the {@link EventLoopGroup} for the parent (acceptor) and the child (client). These
     * {@link EventLoopGroup}'s are used to handle all the events and IO for {@link SocketChannel} and
     * {@link Channel}'s.
     */
    public ServerBootstrap group(EventLoopGroup parentGroup, EventLoopGroup childGroup) {
        super.group(parentGroup);
        if (childGroup == null) {
            throw new NullPointerException("childGroup");
        }
        if (this.childGroup != null) {
            throw new IllegalStateException("childGroup set already");
        }
        this.childGroup = childGroup;
        return this;
    }