Netty初識

話不多說直接從代碼入手

服務器端

NioEventLoopGroup group = new NioEventLoopGroup();//1
try {    
  ServerBootstrap bootstrap = new ServerBootstrap();//2
  bootstrap.group(group,group)            //3
           .channel(NioServerSocketChannel.class) //4           
            .localAddress(new InetSocketAddress(port)) //5           
            .childHandler(new ChannelInitializer<SocketChannel>() {//6                
                  @Override                
                  protected void initChannel(SocketChannel socketChannel) throws Exception {                    
                      socketChannel.pipeline().addLast(new EchoServerHandler());                
                  }           
                 });   
   ChannelFuture channelFuture = bootstrap.bind().sync();    //7
   System.out.println(EchoServer.class.getName() + "started and listen on " + channelFuture);    
    channelFuture.channel().closeFuture().sync();
} finally {    
      group.shutdownGracefully().sync();//8
}

以上是用netty實現(xiàn)最簡單的服務器程序的一段代碼，相比于用NIO實現(xiàn)相應功能，這樣的代碼不能再簡潔了。
下面來分析一下這幾句代碼各自封裝了什么功能：

1、此處聲明了一個EventLoopGroup，顧名思義就是一個eventloop。

查看其繼承關系EventLoopGroup->MultithreadEventLoopGroup ->MultithreadEventExecutorGroup
MultithreadEventExecutorGroup中有以下幾個屬性

private final EventExecutor[] children;
private final Set<EventExecutor> readonlyChildren;
private final AtomicInteger childIndex = new AtomicInteger();
private final AtomicInteger terminatedChildren = new AtomicInteger();
private final Promise<?> terminationFuture = new DefaultPromise(GlobalEventExecutor.INSTANCE);
private final EventExecutorChooser chooser;

其主要屬性即為EventExecutor[] children；
定位到其創(chuàng)建的地方

protected SingleThreadEventExecutor(EventExecutorGroup parent, Executor executor, boolean addTaskWakesUp) {    
super(parent);    
if (executor == null) {       
 throw new NullPointerException("executor");    
}    
this.addTaskWakesUp = addTaskWakesUp;    
this.executor = executor;    
taskQueue = newTaskQueue();
}

這里管理了一個taskQueue用于保存將要執(zhí)行的任務。并在線程中不斷poll隊列中的task并執(zhí)行。

protected Runnable takeTask() {    
assert inEventLoop();    
if (!(taskQueue instanceof BlockingQueue)) {        
throw new UnsupportedOperationException();    
}    
BlockingQueue<Runnable> taskQueue = (BlockingQueue<Runnable>) this.taskQueue;   
 for (;;) {        
ScheduledFutureTask<?> scheduledTask = peekScheduledTask();        
if (scheduledTask == null) {            
Runnable task = null;           
 try {              
  task = taskQueue.take();                
if (task == WAKEUP_TASK) {                    
task = null;               
 }           
 } catch (InterruptedException e) {                
// Ignore         
   }        
    return task;       
 } else {          
  long delayNanos = scheduledTask.delayNanos();          
  Runnable task = null;          
  if (delayNanos > 0) {                
try {                   
 task = taskQueue.poll(delayNanos, TimeUnit.NANOSECONDS);          
      } catch (InterruptedException e) {                    // Waken up.            
        return null;             
   }        
 }          
  if (task == null) {                // We need to fetch the scheduled tasks now as otherwise there may be a chance that                // scheduled tasks are never executed if there is always one task in the taskQueue.                // This is for example true for the read task of OIO Transport                // See https://github.com/netty/netty/issues/1614                fetchFromScheduledTaskQueue();               
 task = taskQueue.poll();           
 }           
 if (task != null) {                
return task;           
 }       
 }    
}}

總而言之，Netty中的EventLoopGroup就是建立了一個EventLoop數(shù)組。并在其中不斷處理新的事務，其中包括selector的輪詢操作和一些用戶自定義的Task。

2、此處申明了一個Bootstrap

查看其代碼，發(fā)現(xiàn)其只有兩個構造函數(shù)，一個無參構造函數(shù)，一個復制構造函數(shù)。

《Netty權威指南》中對其解釋如下:
其根本原因為它的參數(shù)太多了，而且未來也可能會發(fā)生變化，為了解決這個問題，就需要引入Builder模式。

這個類是Netty的一個輔助類，提供方法設置啟動相關的參數(shù)。

3、綁定group

從下面的代碼可以看出應該有兩個group來完成Server端，parent負責acceptor，child作為client，而當其只傳一個group時，這個group需要完成兩件事情。

  /**
     * Specify the {@link EventLoopGroup} which is used for the parent (acceptor) and the child (client).
     */
    @Override
    public ServerBootstrap group(EventLoopGroup group) {
        return group(group, group);
    }

    /**
     * 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 ServerChannel} 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;
    }

4，5、綁定相應的channel，并為其綁定端口

這里通過反射的工廠方法建立了一個NIOServerSocketChannel

    /**
     * The {@link Class} which is used to create {@link Channel} instances from.
     * You either use this or {@link #channelFactory(io.netty.channel.ChannelFactory)} if your
     * {@link Channel} implementation has no no-args constructor.
     */
    public B channel(Class<? extends C> channelClass) {
        if (channelClass == null) {
            throw new NullPointerException("channelClass");
        }
        return channelFactory(new ReflectiveChannelFactory<C>(channelClass));
    }

6、設置處理handle，在pipeline中添加相應的回調函數(shù)

    private class EchoServerHandler extends ChannelHandlerAdapter {
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            super.exceptionCaught(ctx, cause);
        }

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

        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
            super.channelReadComplete(ctx);
        }
    }

7、最后一步綁定本地端口，啟動服務

 private ChannelFuture doBind(final SocketAddress localAddress) {
        final ChannelFuture regFuture = initAndRegister();//用工廠方法創(chuàng)建Channel
        final Channel channel = regFuture.channel();
        if (regFuture.cause() != null) {
            return regFuture;
        }

        if (regFuture.isDone()) {
            // At this point we know that the registration was complete and successful.
            ChannelPromise promise = channel.newPromise();
            doBind0(regFuture, channel, localAddress, promise);
            return promise;
        } else {
            // Registration future is almost always fulfilled already, but just in case it's not.
            final PendingRegistrationPromise promise = new PendingRegistrationPromise(channel);
            regFuture.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {
                    Throwable cause = future.cause();
                    if (cause != null) {
                        // Registration on the EventLoop failed so fail the ChannelPromise directly to not cause an
                        // IllegalStateException once we try to access the EventLoop of the Channel.
                        promise.setFailure(cause);
                    } else {
                        // Registration was successful, so set the correct executor to use.
                        // See https://github.com/netty/netty/issues/2586
                        promise.executor = channel.eventLoop();
                    }
                    doBind0(regFuture, channel, localAddress, promise);
                }
            });
            return promise;
        }
    }

在newChannel后對channel進行初始化，這個方法由ServerBootstrap實現(xiàn)，代碼如下

 @Override
    void init(Channel channel) throws Exception {
        final Map<ChannelOption<?>, Object> options = options();
        synchronized (options) {
            channel.config().setOptions(options);
        }

        final Map<AttributeKey<?>, Object> attrs = attrs();
        synchronized (attrs) {
            for (Entry<AttributeKey<?>, Object> e: attrs.entrySet()) {
                @SuppressWarnings("unchecked")
                AttributeKey<Object> key = (AttributeKey<Object>) e.getKey();//設置socket相應的屬性
                channel.attr(key).set(e.getValue());
            }
        }

        ChannelPipeline p = channel.pipeline();//將AbstractBootstrap的handler添加到NioServerSocektChannel的ChannelPipeline中
        if (handler() != null) {
            p.addLast(handler());
        }

        final EventLoopGroup currentChildGroup = childGroup;
        final ChannelHandler currentChildHandler = childHandler;
        final Entry<ChannelOption<?>, Object>[] currentChildOptions;
        final Entry<AttributeKey<?>, Object>[] currentChildAttrs;
        synchronized (childOptions) {
            currentChildOptions = childOptions.entrySet().toArray(newOptionArray(childOptions.size()));
        }
        synchronized (childAttrs) {
            currentChildAttrs = childAttrs.entrySet().toArray(newAttrArray(childAttrs.size()));
        }

        p.addLast(new ChannelInitializer<Channel>() {
            @Override
            public void initChannel(Channel ch) throws Exception {
                ch.pipeline().addLast(new ServerBootstrapAcceptor(
                        currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));//將注冊的ServerBootstrapAcceptor注冊到pipeline中
            }
        });
    }
最后編輯于 ：
?著作權歸作者所有,轉載或內容合作請聯(lián)系作者
【社區(qū)內容提示】社區(qū)部分內容疑似由AI輔助生成，瀏覽時請結合常識與多方信息審慎甄別。
平臺聲明：文章內容（如有圖片或視頻亦包括在內）由作者上傳并發(fā)布，文章內容僅代表作者本人觀點，簡書系信息發(fā)布平臺，僅提供信息存儲服務。

相關閱讀更多精彩內容

友情鏈接更多精彩內容