Netty线程模型详解，看看Netty是如何实现Reactor的

炎炎录 2021-08-24

616

一、背景

看过之前文章的同学都知道，Netty 是一个异步事件驱动的网络应用程序框架，用于快速开发可维护的高性能协议服务器和客户端。

上一篇文章也聊了Reactor线程模型的设计，主要是偏理论方向的，那么今天我们聊一聊实战，就是netty是如何实现Reactor模式的，一方面帮助我们了解netty的底层实现，另外一方面当我们遇到同样的业务场景时，也可以自己实现一个Reactor模式。

对Reactor不熟悉的，可以看看上一篇：

Netty线程模型架构设计，Reactor线程模型

二、Reactor核心角色

前文得知，netty使用的是主从Reactor多线程模型，话不多说，直接上图：

从图中也可以看出，核心角色：

1、MainReactor、SubReactor：主从Reactor线程，两者结构相似，功能不同。

MainReactor通过select监控建立连接事件，收到事件后通过Acceptor接收，处理建立连接事件，然后把建立好的连接分配给子线程SubReactor处理
SubReactor将连接加入连接队列进行监听，并创建一个Handler用于处理各种连接事件

2、Acceptor：处理客户端连接请求

3、Handler：执行非阻塞读/写，read读取数据后，会分发给后面的Worker线程池进行业务处理

4、Worker：Worker线程池会分配独立的线程完成真正的业务处理，如何将响应结果发给Handler进行处理

主要是以上四块，下面我们看看Netty是如何实现的，这里以netty源码为主，但是只考虑主逻辑，一些异常分支会省略

三、Netty实现的Reactor

如下图：

复制

Boss Group 对应的是 MainReactor 角色，Work Group 对应的是 SubReactor 角色

Boss Group 和 Work Group 都是 NioEventLoopGroup, 只是负责的分工不同。Boss Group 下的 EventLoop 处理 Accept 事件，Work Group 下的 EventLoop 处理 Read / Write 等事件

下面我们详细了解几个很核心的概念

先看看下面 EventLoopGroup、EventLoop 的类图，看个大概就行，类太多，没有理解的时候怕没有耐心看完。

简单总结一下：

EventLoop 继承 EventLoopGroup

最上面是 JDK 的 JUC 包里的 Executor，说明 EventLoop 、EventLoopGroup 都是线程池

EventLoopGroup：

下面直接从代码层面出发，看看是 EventLoopGroup 的结构及工作流程，我们以NioEventLoopGroup 为例。

在 Netty 服务端中，我们会设置的两个组，如下：

复制

this.bossGroup = new NioEventLoopGroup();
this.workerGroup = new NioEventLoopGroup();
复制

先看看NioEventLoopGroup类的构造函数

public NioEventLoopGroup() {
    this(0);


}
public NioEventLoopGroup(int nThreads) {
    this(nThreads, (Executor) null);
}
复制


复制

是一层一层嵌套的，每层都会添加一个默认参数

下图是全参数的构造函数：

public NioEventLoopGroup(int nThreads, Executor executor, EventExecutorChooserFactory chooserFactory,
                         final SelectorProvider selectorProvider,
                         final SelectStrategyFactory selectStrategyFactory,
                         final RejectedExecutionHandler rejectedExecutionHandler) {
    super(nThreads, executor, chooserFactory, selectorProvider, selectStrategyFactory, rejectedExecutionHandler);
}
复制

看到参数这么多，先尝试着简单解释一下：

nThreads: 线程池中的线程数，也就是 NioEventLoop 的实例数量；默认值：CPU * 2
executor: 这个就比较奇怪了，我们本身就是个线程池，为啥还要传一个线程池实例？（后面讲）
chooserFactory：当任务提交时，线程池选择一个线程来执行这个任务，就是选择策略
selectorProvider：selector提供者，因为每个NioEventLoop都有一个selector，所以需要一个provider
selectStrategyFactory：selector策略工厂，每个NioEventLoop都有一个selector，所以需要一个selector的策略工厂，说白了就是工作流程
rejectedExecutionHandler：拒绝策略

这么多参数，具体干嘛用的，一路追查下去，我们找到他的父类的父类，看看构造函数：

复制

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


    if (executor == null) {
        executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());
    }


    children = new EventExecutor[nThreads];
    for (int i = 0; i < nThreads; i ++) {
        boolean success = false;


        try {


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


    chooser = chooserFactory.newChooser(children);
    final FutureListener<Object> terminationListener = new FutureListener<Object>() {
        @Override
        public void operationComplete(Future<Object> future) throws Exception {
            if (terminatedChildren.incrementAndGet() == children.length) {
                terminationFuture.setSuccess(null);
            }
        }
    };
    for (EventExecutor e: children) {
        e.terminationFuture().addListener(terminationListener);
    }


    Set<EventExecutor> childrenSet = new LinkedHashSet<EventExecutor>(children.length);
    Collections.addAll(childrenSet, children);
    readonlyChildren = Collections.unmodifiableSet(childrenSet);
}
复制

仔细看一下这里的逻辑：

a、children 先赋值一个EventExecutor数组，大小就是传递进来的值，若没有传递，默认值为当前CPU * 2

复制

children = new EventExecutor[nThreads];
复制

b、循环实例化children，异常逻辑就不看了

复制

for (int i = 0; i < nThreads; i ++) {
    boolean success = false;
    try {
        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 {
        。。。略
        }
    }
}
复制

具体看看实例化方法，把NioEventLoopGroup带着的参数传递进去

复制

protected EventLoop newChild(Executor executor, Object... args) throws Exception {
    return new NioEventLoop(this, executor, (SelectorProvider) args[0],
        ((SelectStrategyFactory) args[1]).newSelectStrategy(), (RejectedExecutionHandler) args[2]);
}
复制

原来EventLoopGroup的参数，大部分是为例传递给EventLoop

c、选择执行任务的线程

chooser = chooserFactory.newChooser(children);
复制

d、后面就是为每个 EventLoop 加销毁的 listener 和设置 EventLoop 为只读Set

复制

final FutureListener<Object> terminationListener = new FutureListener<Object>() {
    @Override
    public void operationComplete(Future<Object> future) throws Exception {
        if (terminatedChildren.incrementAndGet() == children.length) {
            terminationFuture.setSuccess(null);
        }
    }
};
for (EventExecutor e: children) {
    e.terminationFuture().addListener(terminationListener);
}


Set<EventExecutor> childrenSet = new LinkedHashSet<EventExecutor>(children.length);
Collections.addAll(childrenSet, children);
readonlyChildren = Collections.unmodifiableSet(childrenSet);
复制

到这里EventLoopGroup的流程就差不多了，下面我们通过newChild方法创建一个EventLoop实例

复制

@Override
protected EventLoop newChild(Executor executor, Object... args) throws Exception {
    return new NioEventLoop(this, executor, (SelectorProvider) args[0],
        ((SelectStrategyFactory) args[1]).newSelectStrategy(), (RejectedExecutionHandler) args[2]);
}
复制

EventLoop:

这里我们以NioEventLoop为主

下面是NioEventLoop类自身的主要属性：

复制

private Selector selector;private Selector unwrappedSelector;
private SelectedSelectionKeySet selectedKeys;
private final SelectorProvider provider;


private final AtomicBoolean wakenUp = new AtomicBoolean();
private final SelectStrategy selectStrategy;
private volatile int ioRatio = 50;
private int cancelledKeys;
private boolean needsToSelectAgain;
复制

再看看NioEventLoop的构造方法：

复制

NioEventLoop(NioEventLoopGroup parent, Executor executor, SelectorProvider selectorProvider,
             SelectStrategy strategy, RejectedExecutionHandler rejectedExecutionHandler) {
    super(parent, executor, false, DEFAULT_MAX_PENDING_TASKS, rejectedExecutionHandler);
    if (selectorProvider == null) {
        throw new NullPointerException("selectorProvider");
    }
    if (strategy == null) {
        throw new NullPointerException("selectStrategy");
    }
    provider = selectorProvider;
    final SelectorTuple selectorTuple = openSelector();
    selector = selectorTuple.selector;
    unwrappedSelector = selectorTuple.unwrappedSelector;
    selectStrategy = strategy;
}
复制

父类方法下面看

核心就是selector初始化，具体openSelector方法下面再讲

看完自身的属性和构造函数，我们再去看看他的父类SingleThreadEventLoop构造函数：

复制

复制

protected SingleThreadEventLoop(EventLoopGroup parent, Executor executor,
                                boolean addTaskWakesUp, int maxPendingTasks,
                                RejectedExecutionHandler rejectedExecutionHandler) {
    super(parent, executor, addTaskWakesUp, maxPendingTasks, rejectedExecutionHandler);
    tailTasks = newTaskQueue(maxPendingTasks);
}
复制

这里初始化一个tailTasks,先略过

再看SingleThreadEventLoop的父类SingleThreadEventExecutor构造函数

复制

protected SingleThreadEventExecutor(EventExecutorGroup parent, Executor executor,
                                    boolean addTaskWakesUp, int maxPendingTasks,
                                    RejectedExecutionHandler rejectedHandler) {
    super(parent);
    this.addTaskWakesUp = addTaskWakesUp;
    this.maxPendingTasks = Math.max(16, maxPendingTasks);
    this.executor = ObjectUtil.checkNotNull(executor, "executor");
    taskQueue = newTaskQueue(this.maxPendingTasks);
    rejectedExecutionHandler = ObjectUtil.checkNotNull(rejectedHandler, "rejectedHandler");
}
复制

看到这里简单总结一下EventLoop核心属性：

一个Selector选择器
一个线程安全的LinkedBlockingQueue任务队列
一个单线程的线程池

其它概念先有个大概了解，后续涉及相关文章会详细讲解

Channel: Netty 网络通信的主体，由它负责同对端进行网络通信、注册和数据操作等功能。每个 Channel 会绑定一个 EventLoop，绑定后 Channel 所有的操作，都由该 EventLoop 实例来完成。

channelhandler：具体处理数据的地方

ChannelPipeline：链接各个 ChannelHandler，通过 ChannelHandlerContext传递上下文

2、具体工作流程

先大概看看下面的流程图

核心逻辑就在EventLoop里面，接下来我们以NioEventLoopGroup

为入手点，把EventLoop（NioEventLoop实例）流程梳理一下

既然EventLoop是线程池，那么看一下启动方法execute：

复制

public void execute(Runnable task) {
    if (task == null) {
        throw new NullPointerException("task");
    }
    // 判断是否为当前线程执行的eventLoop
    boolean inEventLoop = inEventLoop();


    if (inEventLoop) {
        // 添加任务到队列, 默认16
        addTask(task);


    } else {
        // 先启动线程
        startThread();   
        // 再添加任务队列     
        addTask(task);
        if (isShutdown() && removeTask(task)) {
            reject();
        }
    }


    if (!addTaskWakesUp && wakesUpForTask(task)) {
        wakeup(inEventLoop);
    }
}
复制

继续看看如何启动线程的

复制

private void startThread() {
    if (STATE_UPDATER.get(this) == ST_NOT_STARTED) {
        // CAS赋值
        if (STATE_UPDATER.compareAndSet(this, ST_NOT_STARTED, ST_STARTED)) {            doStartThread();
        }
    }
}


private void doStartThread() {


    assert thread == null;


    // 线程池创建一个线程实例
    executor.execute(new Runnable() {
        
        @Override
        public void run() {
            // 设置线程为当前线程
            thread = Thread.currentThread();
            if (interrupted) {
                thread.interrupt();
            }


            boolean success = false;
            updateLastExecutionTime();


            try {
                // 执行SingleThreadEventExecutor 的run方法
               SingleThreadEventExecutor.this.run();
                success = true;
            } catch (Throwable t) {
                logger.warn("Unexpected exception from an event executor: ", t);


            } finally {
                ...略
}
        }
    });
}
复制

线程启动后，会执行SingleThreadEventExecutor(NioEventLoop)的run()方法,终于来到核心的地方了，先贴段代码：

protected void run() {
        // 首先是个死循环,处理IO事件和异步任务
        for (;;) {
            try {
                // 根据是否有任务，来判断具体操作
                // 有任务，执行selectNow(),检查当前是否有就绪的 IO 事件, 如果有, 则返回就绪 IO 事件的个数; 如果没有, 则返回0，该方法不会阻塞
                // 没有任务，执行select(wakenUp.getAndSet(false)),会阻塞，下面会详细讲
                switch (selectStrategy.calculateStrategy(selectNowSupplier, hasTasks())) {
                    case SelectStrategy.CONTINUE:
                        continue;
                    case SelectStrategy.SELECT:
                        select(wakenUp.getAndSet(false));


                        if (wakenUp.get()) {
                            selector.wakeup();
                        }
                    default:
                        // fallthrough
                }


                cancelledKeys = 0;
                needsToSelectAgain = false;
                // IO操作与任务处理时间占比，默认为50%，可以在辅助类ServerBootstrap配置启动参数时设置
                // 值越大，处理IO操作占比越大，当为100时，不考虑时间，先处理IO事件，再执行异步任务
                final int ioRatio = this.ioRatio;
                if (ioRatio == 100) {
                    try {
                        // 查询就绪的 IO 事件，下面详细讲
                        processSelectedKeys();
                    } finally {
                        // 任务处理机制,下面详细讲
                        runAllTasks();
                    }
                } else {
                    final long ioStartTime = System.nanoTime();
                    try {
                        processSelectedKeys();
                    } finally {
                        // Ensure we always run tasks.
                        final long ioTime = System.nanoTime() - ioStartTime;
                        runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
                    }
                }
            } catch (Throwable t) {
                handleLoopException(t);
            }
            // Always handle shutdown even if the loop processing threw an exception.
            try {
                if (isShuttingDown()) {
                    closeAll();
                    if (confirmShutdown()) {
                        return;
                    }
                }
            } catch (Throwable t) {
                handleLoopException(t);
            }
        }
    }
复制

到这里，我们终于找到了NioEventLoop事件循环的核心，下面我们再来分析一下（select、processSelectedKeys、runAllTasks）这三个核心的方法。

先看看select源码：

//这个方法解决了Nio中臭名昭著的bug：selector的select方法导致空轮询 cpu100% 
private void select(boolean oldWakenUp) throws IOException {
    Selector selector = this.selector;
    try {
        int selectCnt = 0;
        long currentTimeNanos = System.nanoTime();


        /* delayNanos(currentTimeNanos)：计算延迟任务队列中第一个任务的到期执行时间（即最晚还能延迟多长时间执行），默认返回1s。每个SingleThreadEventExecutor都持有一个延迟执行任务的优先队列PriorityQueue，启动线程时，往队列中加入一个任务。*/
        long selectDeadLineNanos = currentTimeNanos + delayNanos(currentTimeNanos);
        for (;;) {
            /* 如果延迟任务队列中第一个任务的最晚还能延迟执行的时间小于500000纳秒，且selectCnt == 0（selectCnt 用来记录selector.select方法的执行次数和标识是否执行过selector.selectNow()），则执行selector.selectNow()方法并立即返回。*/
            long timeoutMillis = (selectDeadLineNanos - currentTimeNanos + 500000L) / 1000000L;
            if (timeoutMillis <= 0) {
                if (selectCnt == 0) {
                    selector.selectNow();
                    selectCnt = 1;
                }
                break;
            }


            if (hasTasks() && wakenUp.compareAndSet(false, true)) {
                selector.selectNow();
                selectCnt = 1;
                break;
            }


            // 超时阻塞select
            int selectedKeys = selector.select(timeoutMillis);
            selectCnt ++;
            System.out.println(selectCnt);


            // 有事件到来 | 被唤醒 | 有内部任务 | 有定时任务时，会返回
            if (selectedKeys != 0 || oldWakenUp || wakenUp.get() || hasTasks() || hasScheduledTasks()) {
                break;
            }


            long time = System.nanoTime();
            if (time - TimeUnit.MILLISECONDS.toNanos(timeoutMillis) >= currentTimeNanos) {
                // 阻塞超时后没有事件到来，重置selectCnt
                selectCnt = 1;
            } else if (SELECTOR_AUTO_REBUILD_THRESHOLD > 0 &&
                    selectCnt >= SELECTOR_AUTO_REBUILD_THRESHOLD) {
                // Selector重建
                rebuildSelector();
                selector = this.selector;
                // Select again to populate selectedKeys.
                selector.selectNow();
                selectCnt = 1;
                break;
            }
            currentTimeNanos = time;
        }
    } catch (CancelledKeyException e) {
        // Harmless exception - log anyway
    }
}
复制

说一下Nio中臭名昭著的bug：

简单来说就是旧版本的Java NIO在Linux Epoll实现上存在bug，(jvm)Selector.select方法可能在没有任何就绪事件的情况下返回，导致CPU空转，利用率飙升到100%

看看netty的解决策略：

1、根据该BUG的特征，首先侦测该BUG是否发生（空转计数）

2、将问题Selector上注册的Channel转移到新建的Selector上（rebuildSelector）

3、老的问题Selector关闭，使用新建的Selector替换

再看一下processSelectedKeys：

private void processSelectedKeys() {
        if (selectedKeys != null) {
            processSelectedKeysOptimized();
        } else {
            processSelectedKeysPlain(selector.selectedKeys());
        }
    }
复制

根据selectedKeys是否为空来确定调用哪个方法

先看看：

private void processSelectedKeysOptimized() {
        // 迭代 selectedKeys 获取就绪的 IO 事件
        for (int i = 0; i < selectedKeys.size; ++i) {
            final SelectionKey k = selectedKeys.keys[i];
            // null out entry in the array to allow to have it GC'ed once the Channel close
            // See https://github.com/netty/netty/issues/2363
            // 方便GC
            selectedKeys.keys[i] = null;
            
            final Object a = k.attachment();


            if (a instanceof AbstractNioChannel) {
                // 为每个事件都调用 processSelectedKey 来处理它
                processSelectedKey(k, (AbstractNioChannel) a);
            } else {
                @SuppressWarnings("unchecked")
                NioTask<SelectableChannel> task = (NioTask<SelectableChannel>) a;
                processSelectedKey(k, task);
            }


            if (needsToSelectAgain) {
                // null out entries in the array to allow to have it GC'ed once the Channel close
                // See https://github.com/netty/netty/issues/2363
                selectedKeys.reset(i + 1);


                selectAgain();
                i = -1;
            }
        }
    }
复制

其实核心就是遍历所有就绪的IO事件，调用processSelectedKey来处理，再看看processSelectedKey函数

private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
        final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();
        ...
  

        try {
            int readyOps = k.readyOps();
            // 连接事件
            if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
                // 连接就绪事件只需要处理一次就行了，否则后续的 select()操作会一直立刻返回
                int ops = k.interestOps();
                ops &= ~SelectionKey.OP_CONNECT;
                k.interestOps(ops);
                // 最后调用 SocketChannel 的 finishConnect()方法
                unsafe.finishConnect();
            }


            // 可写事件
            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();
            }


            // 可读事件 | accepter 事件
            if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
                unsafe.read();
            }
        } catch (CancelledKeyException ignored) {
            unsafe.close(unsafe.voidPromise());
        }
    }
复制

到这里，如果看过nio selector源码的同学肯定会非常熟悉了，篇幅有限，就不详细介绍每个事件源码了，简单介绍一下事件

ACCEPT事件：连接建立好之后将该连接的channel注册到workGroup中某个NIOEventLoop的selector中
READ事件：从channel中读取数据，存放到byteBuf中，触发后续的ChannelHandler来处理数据
WRITE事件：正常情况下一般是不会注册写事件的，如果Socket发送缓冲区中没有空闲内存时，在写入会导致阻塞，此时可以注册写事件，当有空闲内存（或者可用字节数大于等于其低水位标记）时，再响应写事件，并触发对应回调
CONNECT事件：该事件是client触发的，由主动建立连接这一侧触发的

processSelectedKeysPlain方法和processSelectedKeysOptimized类似，核心都是调用processSelectedKey函数来处理，就不再展开。

最后看看runAllTasks函数，其实有两个方法，一个是有参数的，一个是没有参数的，我们来看看有参数的实现：

protected boolean runAllTasks(long timeoutNanos) {
        // 获取已到期的定时任务
        fetchFromScheduledTaskQueue();
        Runnable task = pollTask();
        if (task == null) {
            afterRunningAllTasks();
            return false;
        }


        final long deadline = ScheduledFutureTask.nanoTime() + timeoutNanos;
        long runTasks = 0;
        long lastExecutionTime;
        for (;;) {
            // 执行任务
            safeExecute(task);


            runTasks ++;
            // 每64个任务检查一次是否超时，相对来说nanoTime()是比较耗时的操作，所以不是每次都检查
            if ((runTasks & 0x3F) == 0) {
                lastExecutionTime = ScheduledFutureTask.nanoTime();
                if (lastExecutionTime >= deadline) {
                    break;
                }
            }
            // 取下一个任务
            task = pollTask();
            if (task == null) {
                lastExecutionTime = ScheduledFutureTask.nanoTime();
                break;
            }
        }
        // 预留的扩展方法
        afterRunningAllTasks();
        this.lastExecutionTime = lastExecutionTime;
        return true;
    }
复制

简单小结 eventLoop 的执行策略：

1、根据hasTasks() 的结果来决定是执行 selectNow() 还是 select()，如果有任务正在等待，那么应该使用无阻塞的 selectNow()，如果没有任务在等待，那么就可以使用带阻塞的 select() 操作

2、IO操作与执行任务的占比，即上述processSelectedKeys()与runAllTasks()方法的执行策略，ioRate是线程分配给 IO 操作所占的时间比(即运行 processSelectedKeys 耗时在整个循环中所占用的时间)

四、总结

1、概念回顾

EventLoop

一个 Selector。
一个任务队列(mpsc_queue: 多生产者单消费者 lock-free)。
一个延迟任务队列(delay_queue: 一个二叉堆结构的优先级队列，复杂度为O(log n))。
EventLoop 绑定了一个 Thread，这直接避免了pipeline 中的线程竞争。

Boss: mainReactor 角色，Worker: subReactor 角色

Boss 和 Worker 共用 EventLoop 的代码逻辑，Boss 处理 accept 事件，Worker 处理 read，write 等事件。
Boss 监听并 accept 连接(channel)后以轮训的方式将 channel 交给 Worker，Worker 负责处理此 channel 后续的read/write 等 IO 事件。
在不 bind 多端口的情况下 BossEventLoopGroup 中只需要包含一个 EventLoop，也只能用上一个，多了没用。
WorkerEventLoopGroup 中一般包含多个 EventLoop，经验值一般为 cpu * 2
Channel 分两大类 ServerChannel 和 Channel，ServerChannel 对应着监听套接字(ServerSocketChannel)，Channel 对应着一个网络连接。

2、源码启发

在大部分场景下，并行多线程处理可以提升系统的并发性能，我们日常工作中也是采用多线程来提升性能，使用锁来保证并发安全。

但是Netty的“无锁化”设计理念，是之前没有考虑过的另一种高性能之道。

简单对比一下两者的区别：

数据分配（多线程+锁）：数据产生后统一放在数据容器中，由数据消费线程自己来获取数据进行处理，这里的获取动作需要保证是安全的，一般通过锁机制来保护，比如Java线程池中线程从阻塞队列中获取任务进行执行，就是由阻塞队列保证线程安全
数据隔离（EventLoop）：数据隔离就是数据产生后就提交给不同的线程来处理，线程内部一般有一个数据容器来保存待处理的数据，这里的提交动作需要保证是安全的，比如Netty的BossGroup将建立好的连接注册到WorkerGroup时，是由内核来保证线程安全的（比如Linux就是通过epoll_ctl方法，该方法是线程安全的）

当然这两者没有说哪种策略就是最好的，软件工程没有银弹，只有更适合的场景，比如数据隔离（EventLoop），没有带来的复杂性，也没有线程切换带来的开销，但是存在一个处理数据不均衡问题。

比如每个channel都绑定了一个EventLoop，那么就肯定会存在，某个EventLoop上面所有绑定的channel多，另外一个EventLoop绑定的channel少，无法做到负载均衡，但是对于数据分配（多线程+锁）来说，就不存在这个问题。

另外还有随处可见的各种经典的设计模式的应用，API接口的封装等等就不细讲了，有兴趣可自己看看源码。

参考：

https://segmentfault.com/a/1190000007403937

https://www.javadoop.com/post/netty-part-7

https://www.cnblogs.com/cherish010/p/9044776.html

另外文中有错误的地方，请帮忙指正，多谢！

数据库

文章转载自炎炎录，如果涉嫌侵权，请发送邮件至：contact@modb.pro进行举报，并提供相关证据，一经查实，墨天轮将立刻删除相关内容。

Netty线程模型详解，看看Netty是如何实现Reactor的

评论

相关阅读