反应器应用的网络编程

写在前面

在netty使用了reactor的线程模型（或者叫做工作模式）。本文就一起来看下其是如何使用的。

1：不同的rector对应的不同的编码方式

首先是rector的单线程模型，对应到netty中的编码方式如下：

// 这里的1，就是rector的单线程模型中一个线程的"1"
NioEventLoopGroup eventExecutors = new NioEventLoopGroup(1);
// 定义启动类，并将rector模型设置到启动类中
ServerBootstrap serverBootstrap = new ServerBootstrap();
serverBootstrap.group(eventExecutors);

非主从多线程版本：

// 这里不设置线程数，线程数会自动根据核数指定（一般多核，所以肯定大于1），就是rector的多线程模型了，当然你可以显式指定线程数量
NioEventLoopGroup eventExecutors = new NioEventLoopGroup();
// 定义启动类，并将rector模型设置到启动类中
ServerBootstrap serverBootstrap = new ServerBootstrap();
serverBootstrap.group(eventExecutors);

主从多线程版本：

// reactor主从模式中的主
NioEventLoopGroup bossGroup = new NioEventLoopGroup();
// reactor主从模式中的从
NioEventLoopGroup workerGroup = new NioEventLoopGroup();
// 定义启动类，并将rector模型设置到启动类中
ServerBootstrap serverBootstrap = new ServerBootstrap();
serverBootstrap.group(bossGroup, workerGroup);

2：netty对reactor主从模式支持源码分析

netty想要支持reactor，需要做的工作其实就是将主reactor绑定到ServerSocketChannel，将从reactor绑定到SocketChannel中就行了，即让主reactor负责基于ServerSocketChannel的接收连接的工作，而让从reactor负责基于SocketChannel的数据读写工作。
debug使用的代码：

// reactor主从模式中的主
NioEventLoopGroup bossGroup = new NioEventLoopGroup();
// reactor主从模式中的从
NioEventLoopGroup workerGroup = new NioEventLoopGroup();
// 定义启动类，并将rector模型设置到启动类中
ServerBootstrap serverBootstrap = new ServerBootstrap();
serverBootstrap.group(bossGroup, workerGroup);

在serverBootstrap.group(bossGroup, workerGroup);中完成绑定的工作。

2.1：主reactor绑定到ServerSocketChannel

// io.netty.bootstrap.ServerBootstrap#group(io.netty.channel.EventLoopGroup, io.netty.channel.EventLoopGroup)
public ServerBootstrap group(EventLoopGroup parentGroup, EventLoopGroup childGroup) {
    // 调用父类group方法
    super.group(parentGroup);
    ...
    return this;
}

// io.netty.bootstrap.AbstractBootstrap#group(io.netty.channel.EventLoopGroup)
public B group(EventLoopGroup group) {
    ...
    // volatile EventLoopGroup group; 设置到局部变量
    this.group = group;
    return self();
}

io.netty.bootstrap.AbstractBootstrap#group()
// 该方法负责读取到上一步设置的局部变量
public final EventLoopGroup group() {
    return group;
}

// io.netty.bootstrap.AbstractBootstrap#initAndRegister
final ChannelFuture initAndRegister() {
    ...
    ChannelFuture regFuture = config().group().register(channel);
    ...
    return regFuture;
}

ChannelFuture regFuture = config().group().register(channel);这里的channel类是netty的channel底层就是Java nio的ServerSocketChannel了，通过register方法也就完成了绑定。

2.2：从reactor绑定到SocketChannel

从reactor绑定到SocketChannel需要依赖于ServerSocketChannel，因为SocketChannel的创建是由ServerSocketChannel来完成的，首先看代码：

// io.netty.bootstrap.ServerBootstrap#group(io.netty.channel.EventLoopGroup, io.netty.channel.EventLoopGroup)
public ServerBootstrap group(EventLoopGroup parentGroup, EventLoopGroup childGroup) {
    ...
    // 赋值从reactor到childGroup
    this.childGroup = childGroup;
    return this;
}

// io.netty.bootstrap.ServerBootstrap#init
void init(Channel channel) throws Exception {
    ...
    // 又换了个名字！
    final EventLoopGroup currentChildGroup = childGroup;
    p.addLast(new ChannelInitializer<Channel>() {
    @Override
    public void initChannel(final Channel ch) throws Exception {
        ...
        ch.eventLoop().execute(new Runnable() {
            @Override
            public void run() {
                // 这里创建ServerBootstrapAcceptor，将从reactor作为参数传了进去
                pipeline.addLast(new ServerBootstrapAcceptor(
                ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
            }
        });
}

// io.netty.bootstrap.ServerBootstrap.ServerBootstrapAcceptor#channelRead
public void channelRead(ChannelHandlerContext ctx, Object msg) {
    // 这里的msg就是socketchannel，所以这里可以强转
    final Channel child = (Channel) msg;
    childGroup.register(child).addListener(new ChannelFutureListener() {
    ...
}

childGroup.register(child)这里就完成绑定了，需要注意，只有在第一次读取客户端数据时才会执行到这里。

3：main reactor为什么只会用到线程组中的一个线程

前面我们分析了main rector是如何帮i当道ServerSocketChannel中的，在如下位置：

可以看到这个绑定是在bind端口时，即代码b.bind(PORT).sync();执行的，因为只会绑定到一个端口，所以使用一个线程也就足够了，多了也没有什么意义。所以只会用到线程组中一个线程的原因是只会绑定一个端口号。所以啊，就要从这一组线程中选出一个线程来，如何选的呢？接着绑定逻辑来看源码：

// io.netty.channel.MultithreadEventLoopGroup#register(io.netty.channel.Channel)
public ChannelFuture register(Channel channel) {
    return next().register(channel);
}

这里的next方法就是来完成选择工作的：

// io.netty.util.concurrent.MultithreadEventExecutorGroup#next
public EventExecutor next() {
    // private final EventExecutorChooserFactory.EventExecutorChooser chooser
    return chooser.next();
}

choose有两个实现类：

private static final class PowerOfTwoEventExecutorChooser implements EventExecutorChooser {
    private final AtomicInteger idx = new AtomicInteger();
    private final EventExecutor[] executors;

    PowerOfTwoEventExecutorChooser(EventExecutor[] executors) {
        this.executors = executors;
    }

    @Override
    public EventExecutor next() {
        // 这种与的方式选择一个效率要由于取余的方式，一般我们都是采用取余的方式，但netty追求极致性能
        // 但要求executors总数是2的次幂（另外要注意，-的优先级高于&，我觉得写成这样子更清晰）
        // executors[idx.getAndIncrement() & (executors.length - 1)]
        return executors[idx.getAndIncrement() & executors.length - 1];
    }
}

private static final class GenericEventExecutorChooser implements EventExecutorChooser {
    private final AtomicInteger idx = new AtomicInteger();
    private final EventExecutor[] executors;

    GenericEventExecutorChooser(EventExecutor[] executors) {
        this.executors = executors;
    }

    @Override
    public EventExecutor next() {
        // 取余选择，效率就要低于位运算的方式了
        return executors[Math.abs(idx.getAndIncrement() % executors.length)];
    }
}

PowerOfTwoEventExecutorChooser采用位运算方式选出一个来效率更高，计算如下：

executors总数4（注意-的优先级高于&）
0 & 4 - 1 = 00000000 & 00000011 = 0（十进制）
1 & 4 - 1 = 00000001 & 00000011 = 1（十进制）
2 & 4 - 1 = 00000010 & 00000011 = 2（十进制）
3 & 4 - 1 = 00000011 & 00000011 = 3（十进制）
4 & 4 - 1 = 00000100 & 00000011 = 0（十进制）
循环了。。。

那么，直接让用户设置一个不就好了吗？为什么还要设置线程组呢？我想这是因为netty为了降低编码的复杂度，从而使得主reactor和从reactor使用相同的编码方式，而底层的差异性就由netty来解决了，所以不得不说netty是一个很优秀的框架啊！

写在后面

参考文章列表

什么是reactor以及其三种版本 。