NIO网络编程（八）—— 多线程优化服务端
多线程架构
之前说到的服务端程序都是在一个线程上进行的，这个线程不仅负责连接客户端发来的请求，同时还要处理读写事件，这样效率还是不够高。如今电脑都是多核处理器，这意味着可以同时进行多个线程，所以服务端应该充分利用这一点。
服务端线程可以建立多个线程，将这些线程分成两组：
- 其中一个线程专门处理 accept 事件，称为Boss线程
- 创建 cpu 核心数的线程，每个线程配一个Selector，轮流处理 read 事件，称为Worker线程
他们之间的的关系可以通过下面这张图进行理解：
Boss线程只负责Accept事件，Worker线程负责客户端与服务端之间的读写问题，他们都各自维护一个Selector负责监听通道的事件。当Boss线程检测到有客户端的连接请求，就会把这个连接返回的SocketChannel注册到某一个Worker线程上。当有读写事件发生时，其中一个Worker线程就会检测到事件，就会在该线程中进行处理，这样的设计做到了功能在线程上的分离。
Worker类的实现
上面分析了Worker类是一个监听并且处理读写事件的新线程，所以在Worker类中需要一个Thread对象用来启动线程，还需要一个Selector用来监听事件管理channel，此外为Worker设定一个name。这些可以在构造函数中进行初始化。
之后需要实现一个register函数，这个函数接收一个SocketChannel对象，Worker类的register函数将这个SocketChannel对象注册到Worker类的selector上。
下面就是完整的Worker类的代码：
static class Worker implements Runnable{
private String name;
private Thread thread;
private Selector selector;
private ConcurrentLinkedQueue<Runnable> queue = new ConcurrentLinkedQueue<>();
public Worker(String name) throws IOException {
this.name = name;
thread = new Thread(this, name);
thread.start();
selector = Selector.open();
}
//初始化线程和selector
public void register(SocketChannel sc) throws IOException {
//将任务添加到队列中
queue.add(()->{
try {
sc.register(this.selector,SelectionKey.OP_READ,null);//boss
} catch (ClosedChannelException e) {
e.printStackTrace();
}
});
selector.wakeup();
}
@Override
public void run() {
while(true){
try{
selector.select();//worker-0 一开始会阻塞，下面执行不了，所以要先wakeup一下来注册
Runnable task = queue.poll();
if(task!=null)
{
task.run();//执行了sc.register(this.selector,SelectionKey.OP_READ,null);//boss
}
Iterator<SelectionKey> iter = this.selector.selectedKeys().iterator();
while (iter.hasNext()) {
SelectionKey key = iter.next();
iter.remove();
if (key.isReadable()) {
ByteBuffer buffer = ByteBuffer.allocate(16);
SocketChannel channel = ((SocketChannel) key.channel());
channel.read(buffer);
buffer.flip();
ByteBufferUtil.debugAll(buffer);//buffer可视化
key.cancel();
}
}
}catch (Exception e)
{
}
}
}
}
在这段代码中，还有一个需要注意的问题：必须保证sc.register(this.selector,SelectionKey.OP_READ,null)不会因为selector.select()被阻塞，否则就不能给socketchannel注册上。所以这里采用的方法是使用消息队列，当有线程想要注册到selector上时，就先使用selector.wakeup()唤醒，紧接着在selector.select()下面进行注册，即拿出消息队列中的注册任务执行。这种方式就能够保证当注册通道时不会被select方法阻塞住。
多Worker
上面讲解了Worker类，在实际服务端代码里，肯定是使用多个Worker去管理多个channel，那么可以使用一个worker数组，每一个channel注册worker数组中的某一个worker，此外，可以使用一个变量去给channel轮流分配worker，具体是这样实现的：
- 创建worker数组，并对每一项进行初始化，此外使用一个workerindex表示给下一个通道分配的worker在数组的下标，需要注意的是，因为是多线程环境，所以不能使用int类型，使用的是AtomicInteger类型。
Worker[] workers = new Worker[5];
for (int i =0 ;i< workers.length;i++)
{
workers[i] = new Worker("worker-"+i);
}
AtomicInteger workerindex = new AtomicInteger();
2、给通道注册worker，这里需要模数组长度以达到循环使用的目的，做到了负载均衡。
workers[workerindex.getAndIncrement()% workers.length].register(sc);
服务端代码
下面给出服务端所有的代码：
@Slf4j
public class ThreadNIOServer {
public static void main(String[] args) {
try(ServerSocketChannel server = ServerSocketChannel.open()) {
Thread.currentThread().setName("BOSS");
server.bind(new InetSocketAddress(8080));
server.configureBlocking(false);
Selector boss = Selector.open();
server.register(boss, SelectionKey.OP_ACCEPT);
//1、创建固定数量的worker
Worker[] workers = new Worker[5];
for (int i =0 ;i< workers.length;i++)
{
workers[i] = new Worker("worker-"+i);
}
AtomicInteger workerindex = new AtomicInteger();
while(true)
{
boss.select();
Set<SelectionKey> selectionKeys = boss.selectedKeys();
Iterator<SelectionKey> iter = selectionKeys.iterator();
while (iter.hasNext())
{
SelectionKey key = iter.next();
iter.remove();
if(key.isAcceptable())
{
SocketChannel sc = server.accept();
sc.configureBlocking(false);
log.debug("connected...{}",sc.getRemoteAddress());
//2、关联socket channel和worker
log.debug("before connect...");
workers[workerindex.getAndIncrement()% workers.length].register(sc);
log.debug("after connect...");
}
else if(key.isReadable())
{
}
}
}
}catch (Exception e){
}
}
static class Worker implements Runnable{
private String name;
private Thread thread;
private Selector selector;
private ConcurrentLinkedQueue<Runnable> queue = new ConcurrentLinkedQueue<>();
public Worker(String name) throws IOException {
this.name = name;
thread = new Thread(this, name);
thread.start();
selector = Selector.open();
}
//初始化线程和selector
public void register(SocketChannel sc) throws IOException {
//将任务添加到队列中
queue.add(()->{
try {
sc.register(this.selector,SelectionKey.OP_READ,null);//boss
} catch (ClosedChannelException e) {
e.printStackTrace();
}
});
selector.wakeup();
}
@Override
public void run() {
while(true){
try{
selector.select();//worker-0 一开始会阻塞，下面执行不了，所以要先wakeup一下来注册
Runnable task = queue.poll();
if(task!=null)
{
task.run();//执行了sc.register(this.selector,SelectionKey.OP_READ,null);//boss
}
Iterator<SelectionKey> iter = this.selector.selectedKeys().iterator();
while (iter.hasNext()) {
SelectionKey key = iter.next();
iter.remove();
if (key.isReadable()) {
ByteBuffer buffer = ByteBuffer.allocate(16);
SocketChannel channel = ((SocketChannel) key.channel());
channel.read(buffer);
buffer.flip();
ByteBufferUtil.debugAll(buffer);//buffer可视化
key.cancel();
}
}
}catch (Exception e)
{
}
}
}
}
}