Netty:EventLoopGroup

标签：

Group：群组，Loop：循环，Event：事件，这几个东西联在一起，相比大家也大概明白它的用途了。

Netty内部都是通过线程在处理各种数据，EventLoopGroup就是用来管理调度他们的，注册Channel，管理他们的生命周期，下面就来看看EventLoopGroup是怎样工作的。

在Netty框架初探中，当我们启动客户端或者服务端时，都要声明一个Group对象

EventLoopGroup group = new NioEventLoopGroup();  

NioEventLoopGroup extends MultithreadEventLoopGroup extends MultithreadEventExecutorGroup

public NioEventLoopGroup() {
	this(0);
}
//他会连续调用内部的构造函数，直到用下面的构造去执行父类的构造
//nThreads此时为0，马上就会提到这个参数的用处
public NioEventLoopGroup(
		int nThreads, Executor executor, final SelectorProvider selectorProvider) {
	super(nThreads, executor, selectorProvider);
}

protected MultithreadEventLoopGroup(int nThreads, ThreadFactory threadFactory, Object... args) {
	super(nThreads == 0 ? DEFAULT_EVENT_LOOP_THREADS : nThreads, threadFactory, args);
}
//nThreads：内部线程数，如果为0，就取默认值，通常我们会设置为处理器个数*2
DEFAULT_EVENT_LOOP_THREADS = Math.max(1, SystemPropertyUtil.getInt(
		"io.netty.eventLoopThreads", Runtime.getRuntime().availableProcessors() * 2));

//这里会根据nThreads创建执行者数组
private final EventExecutor[] children;

protected MultithreadEventExecutorGroup(int nThreads, Executor executor, Object... args) {
	if (nThreads <= 0) {
		throw new IllegalArgumentException(String.format("nThreads: %d (expected: > 0)", nThreads));
	}

	if (executor == null) {
		executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());
	}
//这里创建EventExecutor数组对象
	children = new EventExecutor[nThreads];
	if (isPowerOfTwo(children.length)) {
		chooser = new PowerOfTwoEventExecutorChooser();
	} else {
		chooser = new GenericEventExecutorChooser();
	}
//此处循环children数组，来创建内部的NioEventLoop对象
	for (int i = 0; i < nThreads; i ++) {
		boolean success = false;
		try {
			//newChild是abstract方法，运行期会执行具体的实例对象的重载
			children[i] = newChild(executor, args);
			success = true;
		} catch (Exception e) {
			// TODO: Think about if this is a good exception type
			throw new IllegalStateException("failed to create a child event loop", e);
		} finally {
			//如果没有成功，做关闭处理
			if (!success) {
				for (int j = 0; j < i; j ++) {
					children[j].shutdownGracefully();
				}

				for (int j = 0; j < i; j ++) {
					EventExecutor e = children[j];
					try {
						while (!e.isTerminated()) {
							e.awaitTermination(Integer.MAX_VALUE, TimeUnit.SECONDS);
						}
					} catch (InterruptedException interrupted) {
						// Let the caller handle the interruption.
						Thread.currentThread().interrupt();
						break;
					}
				}
			}
		}
	}

	......
}

    @Override
    protected EventLoop newChild(Executor executor, Object... args) throws Exception {
        return new NioEventLoop(this, executor, (SelectorProvider) args[0]);
    }

NioEventLoop extends SingleThreadEventLoop extends SingleThreadEventExecutor

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

服务端的bind方法执行的是AbstractBootstrap的bind方法，bind方法中先会调用validate()方法检查是否有group

    @SuppressWarnings("unchecked")
    public B validate() {
        if (group == null) {
            throw new IllegalStateException("group not set");
        }
        if (channelFactory == null) {
            throw new IllegalStateException("channel or channelFactory not set");
        }
        return (B) this;
    }

最终bind调用doBind，然后调用doBind0，启动一个Runnable线程

    private static void doBind0(
            final ChannelFuture regFuture, final Channel channel,
            final SocketAddress localAddress, final ChannelPromise promise) {

        // This method is invoked before channelRegistered() is triggered.  Give user handlers a chance to set up
        // the pipeline in its channelRegistered() implementation.
        channel.eventLoop().execute(new Runnable() {
            @Override
            public void run() {
                if (regFuture.isSuccess()) {
                    channel.bind(localAddress, promise).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
                } else {
                    promise.setFailure(regFuture.cause());
                }
            }
        });
    }

我们在创建Bootstrap初期，会调用它的group方法，绑定一个group，这样，一个循环就开始运行了。

Netty:EventLoopGroup

标签：

原文地址：http://blog.csdn.net/bdmh/article/details/49945765