💡 如何拿到 cpu 个数

  • Runtime.getRuntime().availableProcessors() 如果工作在 docker 容器下，因为容器不是物理隔离的，会拿到物理 cpu 个数，而不是容器申请时的个数
  • 这个问题直到 jdk 10 才修复，使用 jvm 参数 UseContainerSupport 配置， 默认开启

分两组选择器

  • 单线程配一个选择器，专门处理 accept 事件
  • 创建 cpu 核心数的线程，每个线程配一个选择器，轮流处理 read 事件
    ```java public class ChannelDemo7 { public static void main(String[] args) throws IOException {
    1.   new BossEventLoop().register();
    }
  1. @Slf4j
  2. static class BossEventLoop implements Runnable {
  3.     private Selector boss;
  4.     private WorkerEventLoop[] workers;
  5.     private volatile boolean start = false;
  6.     AtomicInteger index = new AtomicInteger();
  8.     public void register() throws IOException {
  9.         if (!start) {
  10.             ServerSocketChannel ssc = ServerSocketChannel.open();
  11.             ssc.bind(new InetSocketAddress(8080));
  12.             ssc.configureBlocking(false);
  13.             boss = Selector.open();
  14.             SelectionKey ssckey = ssc.register(boss, 0, null);
  15.             ssckey.interestOps(SelectionKey.OP_ACCEPT);
  16.             workers = initEventLoops();
  17.             new Thread(this, "boss").start();
  18.             log.debug("boss start...");
  19.             start = true;
  20.         }
  21.     }
  23.     public WorkerEventLoop[] initEventLoops() {

// EventLoop[] eventLoops = new EventLoop[Runtime.getRuntime().availableProcessors()]; WorkerEventLoop[] workerEventLoops = new WorkerEventLoop[2]; for (int i = 0; i < workerEventLoops.length; i++) { workerEventLoops[i] = new WorkerEventLoop(i); } return workerEventLoops; }

  1.     @Override
  2.     public void run() {
  3.         while (true) {
  4.             try {
  5.                 boss.select();
  6.                 Iterator<SelectionKey> iter = boss.selectedKeys().iterator();
  7.                 while (iter.hasNext()) {
  8.                     SelectionKey key = iter.next();
  9.                     iter.remove();
  10.                     if (key.isAcceptable()) {
  11.                         ServerSocketChannel c = (ServerSocketChannel) key.channel();
  12.                         SocketChannel sc = c.accept();
  13.                         sc.configureBlocking(false);
  14.                         log.debug("{} connected", sc.getRemoteAddress());
  15.                         workers[index.getAndIncrement() % workers.length].register(sc);
  16.                     }
  17.                 }
  18.             } catch (IOException e) {
  19.                 e.printStackTrace();
  20.             }
  21.         }
  22.     }
  23. }
  25. @Slf4j
  26. static class WorkerEventLoop implements Runnable {
  27.     private Selector worker;
  28.     private volatile boolean start = false;
  29.     private int index;
  31.     private final ConcurrentLinkedQueue<Runnable> tasks = new ConcurrentLinkedQueue<>();
  33.     public WorkerEventLoop(int index) {
  34.         this.index = index;
  35.     }
  37.     public void register(SocketChannel sc) throws IOException {
  38.         if (!start) {
  39.             worker = Selector.open();
  40.             new Thread(this, "worker-" + index).start();
  41.             start = true;
  42.         }
  43.         tasks.add(() -> {
  44.             try {
  45.                 SelectionKey sckey = sc.register(worker, 0, null);
  46.                 sckey.interestOps(SelectionKey.OP_READ);
  47.                 worker.selectNow();
  48.             } catch (IOException e) {
  49.                 e.printStackTrace();
  50.             }
  51.         });
  52.         worker.wakeup();
  53.     }
  55.     @Override
  56.     public void run() {
  57.         while (true) {
  58.             try {
  59.                 worker.select();
  60.                 Runnable task = tasks.poll();
  61.                 if (task != null) {
  62.                     task.run();
  63.                 }
  64.                 Set<SelectionKey> keys = worker.selectedKeys();
  65.                 Iterator<SelectionKey> iter = keys.iterator();
  66.                 while (iter.hasNext()) {
  67.                     SelectionKey key = iter.next();
  68.                     if (key.isReadable()) {
  69.                         SocketChannel sc = (SocketChannel) key.channel();
  70.                         ByteBuffer buffer = ByteBuffer.allocate(128);
  71.                         try {
  72.                             int read = sc.read(buffer);
  73.                             if (read == -1) {
  74.                                 key.cancel();
  75.                                 sc.close();
  76.                             } else {
  77.                                 buffer.flip();
  78.                                 log.debug("{} message:", sc.getRemoteAddress());
  79.                                 debugAll(buffer);
  80.                             }
  81.                         } catch (IOException e) {
  82.                             e.printStackTrace();
  83.                             key.cancel();
  84.                             sc.close();
  85.                         }
  86.                     }
  87.                     iter.remove();
  88.                 }
  89.             } catch (IOException e) {
  90.                 e.printStackTrace();
  91.             }
  92.         }
  93.     }
  94. }

} ```