💡 利用多线程优化

现在都是多核 cpu，设计时要充分考虑别让 cpu 的力量被白白浪费

前面的代码只有一个选择器，没有充分利用多核 cpu，如何改进呢？
分两组选择器

  • 单线程配一个选择器，专门处理 accept 事件
  • 创建 cpu 核心数的线程，每个线程配一个选择器，轮流处理 read 事件

1、第一种多线程实现方式

思路 ： 主线程监听接收事件，主线程将客户端通道注册 到worker线程选择器，并设置监听可读事件，由worker线程进行读操作
> 问题：> 主线程的> clientChannel.register(worker.selector,SelectionKey.OP_READ);> 语句和worker线程的> selector.select();> 执行顺序问题 一旦，worker线程陷入阻塞，那么客户端通道注册可读事件无效，worker线程无法感知到读事件

  1. @Slf4j
  2. public class MultiThreadServer {
  4.     public static void main(String[] args) throws IOException {
  5.         Thread.currentThread().setName("boss");
  6.         ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
  7.         serverSocketChannel.configureBlocking(false);
  8.         serverSocketChannel.bind(new InetSocketAddress(8989));
  10.         Selector bossSelector = Selector.open();
  11.         SelectionKey bossKey = serverSocketChannel.register(bossSelector, SelectionKey.OP_ACCEPT);
  12.         //创建固定数量的worker
  13.         Worker worker = new Worker("worker-0");
  14.         worker.register();
  15.         while (true){
  16.             bossSelector.select();
  17.             Iterator<SelectionKey> iterator = bossSelector.selectedKeys().iterator();
  18.             while (iterator.hasNext()){
  19.                 SelectionKey key = iterator.next();
  20.                 iterator.remove();
  21.                 if (key.isAcceptable()) {
  22.                     SocketChannel clientChannel = serverSocketChannel.accept();
  23.                     clientChannel.configureBlocking(false);
  24.                     log.info("connected...{}",clientChannel.getRemoteAddress());
  25.                     //关联worker的selector
  26.                     log.info("before register...{}",clientChannel.getRemoteAddress());
  27.                     clientChannel.register(worker.selector,SelectionKey.OP_READ);
  28.                     log.info("after register...{}",clientChannel.getRemoteAddress());
  29.                 }
  30.             }
  31.         }
  32.     }
  34.     static class Worker implements Runnable{
  35.         private Thread thread;
  36.         private Selector selector;
  37.         private String name;
  39.         private volatile boolean start = false;
  41.         public Worker(String name) {
  42.             this.name = name;
  43.         }
  45.         public synchronized void register() throws IOException {
  46.             if(!start){
  47.                 thread = new Thread(this);
  48.                 selector = Selector.open();
  49.                 start = true;
  50.                 thread.start();
  51.             }
  52.         }
  54.         @Override
  55.         public void run() {
  56.             while (true){
  57.                 try {
  58.                     //选择器阻塞时，无法反映通道注册时间
  59.                     selector.select();
  60.                     Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
  61.                     while (iterator.hasNext()){
  62.                         SelectionKey key = iterator.next();
  63.                         iterator.remove();
  65.                         if (key.isReadable()) {
  66.                             ByteBuffer buffer = ByteBuffer.allocate(1024);
  67.                             SocketChannel clientChannel = (SocketChannel) key.channel();
  68.                             log.info("read...{}",clientChannel.getRemoteAddress());
  69.                             clientChannel.read(buffer);
  70.                             buffer.flip();
  71.                             debugAll(buffer);
  72.                         }
  73.                     }
  74.                 } catch (IOException e) {
  75.                     e.printStackTrace();
  76.                 }
  77.             }
  78.         }
  79.     }
  80. }

2、使用队列解决第一种方式出现的问题

把注册选择器事件通过队列也交给worker线程处理，然后唤醒selector，从队列取出注册事件执行

  1. public class MultiThreadServer {
  3.     public static void main(String[] args) throws IOException {
  4.         Thread.currentThread().setName("boss");
  5.         ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
  6.         serverSocketChannel.configureBlocking(false);
  7.         serverSocketChannel.bind(new InetSocketAddress(8989));
  9.         Selector bossSelector = Selector.open();
  10.         SelectionKey bossKey = serverSocketChannel.register(bossSelector, SelectionKey.OP_ACCEPT);
  11.         //创建固定数量的worker
  12.         Worker worker = new Worker("worker-0");
  13.         while (true){
  14.             bossSelector.select();
  15.             Iterator<SelectionKey> iterator = bossSelector.selectedKeys().iterator();
  16.             while (iterator.hasNext()){
  17.                 SelectionKey key = iterator.next();
  18.                 iterator.remove();
  19.                 if (key.isAcceptable()) {
  20.                     SocketChannel clientChannel = serverSocketChannel.accept();
  21.                     clientChannel.configureBlocking(false);
  22.                     log.info("connected...{}",clientChannel.getRemoteAddress());
  23.                     //关联worker的selector
  24.                     log.info("before register...{}",clientChannel.getRemoteAddress());
  25.                     worker.register(clientChannel);
  26.                     log.info("after register...{}",clientChannel.getRemoteAddress());
  27.                 }
  28.             }
  29.         }
  30.     }
  32.     static class Worker implements Runnable{
  33.         private Thread thread;
  34.         private Selector selector;
  35.         private String name;
  37.         private ConcurrentLinkedDeque<Runnable> queue = new ConcurrentLinkedDeque();
  39.         private volatile boolean start = false;
  41.         public Worker(String name) {
  42.             this.name = name;
  43.         }
  45.         /**
  46.          * 第一次调用会创建work线程 每次注册新事件后唤醒selector
  47.          * @param clientChannel
  48.          * @throws IOException
  49.          */
  50.         public synchronized void register(SocketChannel clientChannel) throws IOException {
  51.             if(!start){
  52.                 thread = new Thread(this);
  53.                 selector = Selector.open();
  54.                 start = true;
  55.                 thread.start();
  56.             }
  57.             //通过队列传递任务，让worker线程自己注册
  58.             queue.add(() -> {
  59.                 try {
  60.                     clientChannel.register(selector,SelectionKey.OP_READ);
  61.                 } catch (ClosedChannelException e) {
  62.                     e.printStackTrace();
  63.                 }
  64.             });
  65.             //唤醒selector
  66.             selector.wakeup();
  67.         }
  69.         @Override
  70.         public void run() {
  71.             while (true){
  72.                 try {
  73.                     selector.select();
  74.                     Runnable task = queue.poll();
  75.                     if(task != null){
  76.                         task.run();
  77.                     }
  78.                     Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
  79.                     while (iterator.hasNext()){
  80.                         SelectionKey key = iterator.next();
  81.                         iterator.remove();
  83.                         if (key.isReadable()) {
  84.                             ByteBuffer buffer = ByteBuffer.allocate(1024);
  85.                             SocketChannel clientChannel = (SocketChannel) key.channel();
  86.                             log.info("read...{}",clientChannel.getRemoteAddress());
  87.                             clientChannel.read(buffer);
  88.                             buffer.flip();
  89.                             System.out.println(StandardCharsets.UTF_8.decode(buffer));
  90.                             buffer.clear();
  91.                         }
  92.                     }
  93.                 } catch (IOException e) {
  94.                     e.printStackTrace();
  95.                 }
  96.             }
  97.         }
  98.     }
  99. }