概述
Netty是一个高性能网络通信框架，其高性能和采用的线程模型密切相关。Netty采用Reactor模式，Reactor 线程模型有三种：Reactor单线程模型、Reactor多线程模型、主从Reactor多线程模型。Netty可以在启动时通过配置灵活的支持这三种线程模型。
传统服务设计
accept方法会阻塞请求，需要为每个请求分配线程进行处理，如果并发量大会耗尽系统资源，导致系统崩溃。
class Server implements Runnable {
public void run() {
try {
ServerSocket ss = new ServerSocket(PORT);
while (!Thread.interrupted())
new Thread(new Handler(ss.accept())).start();
// or, single-threaded, or a thread pool
} catch (IOException ex) { /* ... */ }
}
static class Handler implements Runnable {
final Socket socket;
Handler(Socket s) {
socket = s;
}
public void run() {
try {
byte[] input = new byte[MAX_INPUT];
socket.getInputStream().read(input);
byte[] output = process(input);
socket.getOutputStream().write(output);
} catch (IOException ex) { /* ... */ }
}
private byte[] process(byte[] cmd) { /* ... */ }
}
}
Reactor模式
Reactor模式又叫反应堆模式，是基于事件驱动的线程模型。使用IO多路复用模型，Reactor线程通过select监听事件，监听到事件后通过dispatch进行分发，如果是连接事件则通过acceptor处理，如果是读写事件则通过对应的handler进行处理。
Reactor单线程模型
所有的连接与读写事件都由Reactor线程进行分发处理，如果某个处理阻塞会导致其他连接请求无法处理。
class Reactor implements Runnable {
final Selector selector;
final ServerSocketChannel serverSocket;
Reactor(int port) throws IOException {
selector = Selector.open();
serverSocket = ServerSocketChannel.open();
serverSocket.socket().bind(new InetSocketAddress(port));
serverSocket.configureBlocking(false);
SelectionKey sk = serverSocket.register(selector, SelectionKey.OP_ACCEPT);
sk.attach(new Acceptor());
}
public void run() {
{
while (!Thread.interrupted()) {
selector.select();
Set selected = selector.selectedKeys();
Iterator it = selected.iterator();
while (it.hasNext()) dispatch((SelectionKey) (it.next()));
selected.clear();
}
} catch(IOException ex){ /* ... */ }
}
void dispatch(SelectionKey k) {
Runnable r = (Runnable) (k.attachment());
if (r != null) r.run();
}
class Acceptor implements Runnable {
public void run() {
try {
SocketChannel c = serverSocket.accept();
if (c != null) new Handler(selector, c);
} catch (IOException ex) { /* ... */ }
}
}
}
final class Handler implements Runnable {
final SocketChannel socket;
final SelectionKey sk;
ByteBuffer input = ByteBuffer.allocate(MAXIN);
ByteBuffer output = ByteBuffer.allocate(MAXOUT);
static final int READING = 0, SENDING = 1;
int state = READING;
Handler(Selector sel, SocketChannel c) throws IOException {
socket = c;
c.configureBlocking(false);
sk = socket.register(sel, 0);
sk.attach(this);
sk.interestOps(SelectionKey.OP_READ);
sel.wakeup();
}
boolean inputIsComplete() { /* ... */ }
boolean outputIsComplete() { /* ... */ }
void process() { /* ... */ }
public void run() {
try {
if (state == READING) read();
else if (state == SENDING) send();
} catch (IOException ex) { /* ... */ }
}
void read() throws IOException {
socket.read(input);
if (inputIsComplete()) {
process();
state = SENDING;
sk.interestOps(SelectionKey.OP_WRITE);
}
}
void send() throws IOException {
socket.write(output);
if (outputIsComplete()) sk.cancel();
}
}
Reactor多线程模型
Reactor线程监听到连接事件则分发给acceptor进行处理，如果不是连接事件则分发给handler进行处理，handler读取数据后交给线程池处理具体的业务逻辑，等处理完成后再由handler将结果发送给客户端。
class Handler implements Runnable {
// uses util.concurrent thread pool
static PooledExecutor pool = new PooledExecutor(...);
static final int PROCESSING = 3;
// ...
synchronized void read() { // ...
socket.read(input);
if (inputIsComplete()) {
state = PROCESSING;
pool.execute(new Processer());
}
}
synchronized void processAndHandOff() {
process();
state = SENDING; // or rebind attachment
sk.interest(SelectionKey.OP_WRITE);
}
class Processer implements Runnable {
public void run() {
processAndHandOff();
}
}
}
主从Reactor多线程模型
连接与响应事件分离，mainReactor只负责处理连接事件，处理完成后将连接绑定到subReactor中的select上，由subReactor负责监听后续事件进行处理。
Selector[] selectors;// also create threads
int next = 0;
class Acceptor { // ...
public synchronized void run() { ...
Socket connection = serverSocket.accept();
if (connection != null)
new Handler(selectors[next], connection);
if (++next == selectors.length)
next = 0;
}
}
Netty线程模型
Netty启动时会实例化两个线程池组：bossGroup、workGroup，bossGroup负责处理客户端连接事件，workGroup负责处理读写事件。
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();
NioEventLoopGroup中包含一个或多个eventLoop，eventLoop是一个不断循环的线程，每个eventLoop内部都有一个selector，用于监听事件。
bossGroup中的eventLoop监听到连接事件并处理后，生成NioSocketChannel并将其注册到workerGroup中的某个eventLoop对应的selector上，并由其负责监听处理读写事件。
