netty--NioEventLoop滴干活

netty是最近項目要用到的nio框架，找了各種資料，發現稱贊它的有點多，所以決定用它：其實也就二選一嘛，mina或netty或自己寫。對於mina，也不熟，不過看各種介紹，貌似netty干活還是很不錯的，尤其是最新的4.x和5.x重構后，且使用結構清晰就先了解了解了。

---

首先要把應用跑起來啦（官網的例子比較多），我這是一個關於mqtt的一個例子：

 

m_bossGroup = new NioEventLoopGroup();
        m_workerGroup = new NioEventLoopGroup();
        
        final NettyMQTTHandler handler = new NettyMQTTHandler();
        handler.setMessaging(messaging);

        ServerBootstrap b = new ServerBootstrap();
            b.group(m_bossGroup, m_workerGroup)
             .channel(NioServerSocketChannel.class) 
             .childHandler(new ChannelInitializer<SocketChannel>() { 
                 @Override
                 public void initChannel(SocketChannel ch) throws Exception {
                    ChannelPipeline pipeline = ch.pipeline();
                    //pipeline.addFirst("metrics", new BytesMetricsHandler(m_metricsCollector));
                    pipeline.addFirst("idleStateHandler", new IdleStateHandler(0, 0, Constants.DEFAULT_CONNECT_TIMEOUT));
                    pipeline.addAfter("idleStateHandler", "idleEventHandler", new MoquetteIdleTimoutHandler());
                    //pipeline.addLast("logger", new LoggingHandler("Netty", LogLevel.ERROR));
                    pipeline.addLast("decoder", new MQTTDecoder());
                    pipeline.addLast("encoder", new MQTTEncoder());
                    pipeline.addLast("metrics", new MessageMetricsHandler(m_metricsCollector));
                    pipeline.addLast("handler", handler);
                 }
             })
             .option(ChannelOption.SO_BACKLOG, 128)
             .option(ChannelOption.SO_REUSEADDR, true)
             .childOption(ChannelOption.SO_KEEPALIVE, true); 
        try {    
            // Bind and start to accept incoming connections.
            ChannelFuture f = b.bind(Constants.PORT);
            LOG.info("Server binded");
            f.sync();
        } catch (InterruptedException ex) {
            LOG.error(null, ex);
        }

再回想下，我們自己寫serversocket的時候是怎么寫的呢（這是一個笨拙的實例代碼）：

    ServerSocket socket;
        channel = ServerSocketChannel.open(); // 打開通道 
        socket = channel.socket(); //得到與通到相關的socket對象 
        socket.bind(new InetSocketAddress(port)); //將scoket榜定在制定的端口上 
        //配置通到使用非阻塞模式，在非阻塞模式下，可以編寫多道程序同時避免使用復雜的多線程 
        channel.configureBlocking(false);
        channel.register(selector, SelectionKey.OP_ACCEPT);
        try {
            while (true) {
                this.selector.select();
                Iterator<SelectionKey> iter = this.selector.selectedKeys().iterator();
                while (iter.hasNext()) {
                    SelectionKey key = iter.next();
                    iter.remove();
                    this.handleKey(key);

                }
            }
        } catch (IOException ex) {
            ex.printStackTrace();
        }

原理還是那些，channel.open(),然后register key，然后遍歷，再然后才進行handleKey()的干活。

那netty的寫法為什么那么瀟灑呢，懷着這個莫名的疑問，我先不管它的結構什么的，直接進行search，發現了這么個東東：

  NioEventLoop(NioEventLoopGroup parent, ThreadFactory threadFactory, SelectorProvider selectorProvider) {
        super(parent, threadFactory, false);
        if (selectorProvider == null) {
            throw new NullPointerException("selectorProvider");
        }
        provider = selectorProvider;
        selector = openSelector();
    }

其中第8行從名稱上來看，有點點意思了，往下看：

 private Selector openSelector() {
        final Selector selector;
        try {
            selector = provider.openSelector();

其中的provider就是我們熟悉的:java.nio.channels.spi.SelectorProvider類。

所以這個就是做了selector.open的工作。

接下來能看到NioEventLoop：

    protected void run() {
        for (;;) {
            oldWakenUp = wakenUp.getAndSet(false);
            try {
                if (hasTasks()) {
                    selectNow();
                } else {
                    select();

再繼續看，該類中處理的selectedKey：

        final NioUnsafe unsafe = ch.unsafe();
        if (!k.isValid()) {
            // close the channel if the key is not valid anymore
            unsafe.close(unsafe.voidPromise());
            return;
        }

        try {
            int readyOps = k.readyOps();
            if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
                unsafe.read();
                if (!ch.isOpen()) {
                    // Connection already closed - no need to handle write.
                    return;
                }
            }
            if ((readyOps & SelectionKey.OP_WRITE) != 0) {
                // Call forceFlush which will also take care of clear the OP_WRITE once there is nothing left to write
                ch.unsafe().forceFlush();
            }
            if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
                // remove OP_CONNECT as otherwise Selector.select(..) will always return without blocking
                // See https://github.com/netty/netty/issues/924
                int ops = k.interestOps();
                ops &= ~SelectionKey.OP_CONNECT;
                k.interestOps(ops);

                unsafe.finishConnect();
            }
        } catch (CancelledKeyException e) {
            unsafe.close(unsafe.voidPromise());
        }
    

現在明白了吧，其實netty也是走這么一套邏輯。

然后再網上看，邏輯是這樣：

NioEventLoopGroup extends MultithreadEventExecutorGroup，其初始化了n個單線程的線程池(children = new SingleThreadEventExecutor[nThreads];)

每個單線程的對象child[i]=NioEventLoop對象，每個NioEventLoop有一個Selector字段。

其run方法是該group都需要干活的具體業務邏輯代碼。

后續再加上別的類說明。