Netty4之EventLoop浅析_JAVA_编程开发_程序员俱乐部

中国优秀的程序员网站程序员频道CXYCLUB技术地图
热搜:
更多>>
 
您所在的位置: 程序员俱乐部 > 编程开发 > JAVA > Netty4之EventLoop浅析

Netty4之EventLoop浅析

 2013/12/10 12:26:11  Tiger_WILL  程序员俱乐部  我要评论(0)
  • 摘要:EventLoopEventLoop继承自EventExecutor,EventLoopGroup。从这点可以看出EventLoop首先是一个Executor,任务执行器。再者它本身是一个环,能够负载均衡执行压力。可以说EventLoop是Netty4的一个核心,整个Netty4的运转都是围绕着它。一、认识EventLoop有哪些方法(下面列举的是一些比较典型的)EventLoopnext()Future<?>submit(Runnabletask
  • 标签:net
EventLoop
EventLoop继承自EventExecutorEventLoopGroup。从这点可以看出EventLoop首先是一个Executor,任务执行器。
再者它本身是一个环,能够负载均衡执行压力。可以说EventLoop是Netty4的一个核心,整个Netty4的运转都是围绕着它。

一、认识EventLoop有哪些方法(下面列举的是一些比较典型的)
class="java"> EventLoop next()
	Future<?> submit(Runnable task)
	ChannelFuture register(Channel channel)
	ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit)
	Future<?> shutdownGracefully()

从这些方法,我们可以清晰的认识到EventLoop可以实现以下功能:
引用 1. 执行任务(IO/非IO任务)
2. 将Channel绑定到EventLoop上
3. 执行定时任务
4. 关闭任务

二、SingleThreadEventLoop & NioEventLoop - EventLoop的2个关键实现
SingleThreadEventLoop继承自SingleThreadEventExecutor这是一个标准的线程池的实现。和JDK中线程池的实现大同小异。主要的用处就是执行任务。
NioEventLoop继承自SingleThreadEventLoop也就是说他具备执行任务的能力。但是它更重要的是每个NioEventLoop里面包含一个Selector,也就是说
它具备网络处理能力。我们来看看它是如何工作的。

  • 启动时打开一个Selector
  • 从下面代码可以看到,Channel是这样绑定到NioEventLoop上的
  • protected void doRegister() throws Exception {
            boolean selected = false;
            for (;;) {
                try {
                    selectionKey = javaChannel().register(eventLoop().selector, 0, this);
                    return;
                } catch (CancelledKeyException e) {
                    if (!selected) {
                        // Force the Selector to select now as the "canceled" SelectionKey may still be
                        // cached and not removed because no Select.select(..) operation was called yet.
                        eventLoop().selectNow();
                        selected = true;
                    } else {
                        // We forced a select operation on the selector before but the SelectionKey is still cached
                        // for whatever reason. JDK bug ?
                        throw e;
                    }
                }
            }
        }
    

  • 从下面代码看到NioEventLoop做的事情就是处理各种NIO事件
  • @Override
        protected void run() {
            for (;;) {
                oldWakenUp = wakenUp.getAndSet(false);
                try {
                    if (hasTasks()) {
                        selectNow();
                    } else {
                        select();
                        if (wakenUp.get()) {
                            selector.wakeup();
                        }
                    }
    
                    cancelledKeys = 0;
    
                    final long ioStartTime = System.nanoTime();
                    needsToSelectAgain = false;
                    if (selectedKeys != null) {
                        processSelectedKeysOptimized(selectedKeys.flip());
                    } else {
                        processSelectedKeysPlain(selector.selectedKeys());
                    }
                    final long ioTime = System.nanoTime() - ioStartTime;
    
                    final int ioRatio = this.ioRatio;
                    runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
    
                    if (isShuttingDown()) {
                        closeAll();
                        if (confirmShutdown()) {
                            break;
                        }
                    }
                } catch (Throwable t) {
                    logger.warn("Unexpected exception in the selector loop.", t);
    
                    // Prevent possible consecutive immediate failures that lead to
                    // excessive CPU consumption.
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        // Ignore.
                    }
                }
            }
        }
    

    private static void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
            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();
                // Also check for readOps of 0 to workaround possible JDK bug which may otherwise lead
                // to a spin loop
                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());
            }
        }
    

  • Read操作最终会当作一个定时任务添加到NioEventLoop中被执行
  • public ChannelHandlerContext read() {
            final DefaultChannelHandlerContext next = findContextOutbound();
            EventExecutor executor = next.executor();
            if (executor.inEventLoop()) {
                next.invokeRead();
            } else {
                Runnable task = next.invokeReadTask;
                if (task == null) {
                    next.invokeReadTask = task = new Runnable() {
                        @Override
                        public void run() {
                            next.invokeRead();
                        }
                    };
                }
                executor.execute(task);
            }
    
            return this;
        }
    

  • 同样的Write操作会当作一个定时任务添加到NioEventLoop中被执行
  • private void write(Object msg, boolean flush, ChannelPromise promise) {
    
            DefaultChannelHandlerContext next = findContextOutbound();
            EventExecutor executor = next.executor();
            if (executor.inEventLoop()) {
                next.invokeWrite(msg, promise);
                if (flush) {
                    next.invokeFlush();
                }
            } else {
                int size = channel.estimatorHandle().size(msg);
                if (size > 0) {
                    ChannelOutboundBuffer buffer = channel.unsafe().outboundBuffer();
                    // Check for null as it may be set to null if the channel is closed already
                    if (buffer != null) {
                        buffer.incrementPendingOutboundBytes(size);
                    }
                }
                safeExecute(executor, WriteTask.newInstance(next, msg, size, flush, promise), promise, msg);
            }
        }
    

    总结:
    • NioEventLoop是Netty4最基础的一个东西
    • 在Netty4里面一切操作都是在NioEventLoop中被执行
    • NioEventLoop既能执行IO任务,也能执行非IO任务,它可以被Channel绑定,处理Channel中IO事物,也可以用来向它提交任务,当作一个任务管理器使用
    发表评论
    用户名: 匿名