Selector多线程selector多线程第一版第二版第三版单线程可以配合selector完成对多个Channel的可读写事件的监控。如果是多核CPU仅仅使用单线程的话就无法充分利用多核CPU的优势。另外只使用单线程如果某个channel上处理事件的耗时太长也会影响其他channel上事件的处理。selector多线程一个线程对应一个selector;一个线程selector处理accept事件建立连接其他线程selector处理其他数据读写事件第一版服务器端主线程publicstaticvoidmain(String[]args)throwsIOException{Thread.currentThread().setName(Boss);ServerSocketChannelserverSocketChannelServerSocketChannel.open();serverSocketChannel.bind(newInetSocketAddress(9527));serverSocketChannel.configureBlocking(false);SelectorselectorSelector.open();SelectionKeyselectionKeyserverSocketChannel.register(selector,0,null);selectionKey.interestOps(SelectionKey.OP_ACCEPT);// 创建固定数量的workerWorkerworkernewWorker(worker-0);worker.init();while(true){selector.select();IteratorSelectionKeyiteratorselector.selectedKeys().iterator();while(iterator.hasNext()){SelectionKeykeyiterator.next();iterator.remove();// 主线程只负责处理 accept事件if(key.isAcceptable()){SocketChannelsocketChannelserverSocketChannel.accept();socketChannel.configureBlocking(false);log.info(客户端已连接socketChannel);log.info(客户端channel注册到worker线程的 selector-开始);// 连接建立后将channel注册到worker线程的 selector上, 并且监听read事件// 【这行代码运行在boss线程】socketChannel.register(worker.selector,SelectionKey.OP_READ,null);log.info(客户端channel注册到worker线程的 selector-结束);}}}}worker线程// worker线程负责处理 读写事件staticclassWorkerimplementsRunnable{// worker线程应该有自己的SelectorprivateSelectorselector;privateThreadthread;privateStringname;privatevolatilebooleanstart;publicWorker(Stringname){this.namename;}// 初始化线程和selectorpublicvoidinit()throwsIOException{// 初始化只做一遍if(!start){threadnewThread(this,name);selectorSelector.open();thread.start();starttrue;}}Overridepublicvoidrun(){// worker线程负责处理 读写事件while(true){try{// 【该行代码运行在worker线程】selector.select();IteratorSelectionKeyiteratorselector.selectedKeys().iterator();while(iterator.hasNext()){SelectionKeyselectionKeyiterator.next();iterator.remove();if(selectionKey.isReadable()){ByteBufferbyteBufferByteBuffer.allocate(16);SocketChannelchannel(SocketChannel)selectionKey.channel();channel.read(byteBuffer);byteBuffer.flip();CharBufferdecodeCharset.defaultCharset().decode(byteBuffer);log.info(收到客户端数据decode);byteBuffer.clear();}}}catch(IOExceptione){thrownewRuntimeException(e);}}}}客户端客户端连接到服务器后立即发送数据先触发了服务器连接事件再出发了read事件。publicstaticvoidmain(String[]args)throwsIOException{SocketChannelsocketChannelSocketChannel.open();socketChannel.connect(newInetSocketAddress(localhost,9527));socketChannel.write(Charset.defaultCharset().encode(1234567890abcdef));System.in.read();}存在问题服务器端打印输出C:\Program Files\Java\jdk-1.8\bin\java.exe ... Connected to the target VM, address: 127.0.0.1:63211, transport: socket 22:29:39 [INFO ] [Boss] c.w.p.SelectorMultiThread - 客户端已连接java.nio.channels.SocketChannel[connected local/127.0.0.1:9527 remote/127.0.0.1:63544] 22:29:39 [INFO ] [Boss] c.w.p.SelectorMultiThread - 客户端channel注册到worker线程的 selector-开始服务器端和客户端分别启动后可以看到客户端成功连接到服务器但是客户端channel没有注册到worker线程的 selector服务器也没有收到客户端的数据没有触发read事件问题原因服务器端创建好worker线程后就直接初始化了worker线程启动后调用了selector的select方法阻塞住了worker线程而worker线程的 selector阻塞住之后就会影响客户端channel的注册因此解决问题的关键就是【先将客户端channel注册到worker线程的selector上再调用worker的selector的select方法】。【客户端channel注册到worker线程的selector上】的逻辑运行在主线程中【调用worker的selector的select方法】的逻辑运行在worker线程中以上两段逻辑的先后顺序如何保证。思考以下方案将worker线程的初始化和【客户端channel注册到worker线程的selector】这两个逻辑放在一起是否可行log.info(客户端channel注册到worker线程的 selector-开始);worker.init();socketChannel.register(worker.selector,SelectionKey.OP_READ,null);log.info(客户端channel注册到worker线程的 selector-结束);根据结果来看似乎可行但是这种方式只是碰巧客户端channel注册到worker线程的selector然后再调用了select方法不能真正的保证顺序而且此时如果再来个客户端此时selector正阻塞在select方法一定会影响新客户端channel的注册。22:53:47 [INFO ] [Boss] c.w.p.SelectorMultiThread - 客户端已连接java.nio.channels.SocketChannel[connected local/127.0.0.1:9527 remote/127.0.0.1:64486] 22:53:47 [INFO ] [Boss] c.w.p.SelectorMultiThread - 客户端channel注册到worker线程的 selector-开始 22:53:47 [INFO ] [Boss] c.w.p.SelectorMultiThread - 客户端channel注册到worker线程的 selector-结束 22:53:47 [INFO ] [worker-0] c.w.p.SelectorMultiThread - 收到客户端数据1234567890abcdef新启动一个客户端可以看到新客户端也成功连接到服务器但是没有注册到worker线程的selector上。22:53:47 [INFO ] [Boss] c.w.p.SelectorMultiThread - 客户端已连接java.nio.channels.SocketChannel[connected local/127.0.0.1:9527 remote/127.0.0.1:64486] 22:53:47 [INFO ] [Boss] c.w.p.SelectorMultiThread - 客户端channel注册到worker线程的 selector-开始 22:53:47 [INFO ] [Boss] c.w.p.SelectorMultiThread - 客户端channel注册到worker线程的 selector-结束 22:53:47 [INFO ] [worker-0] c.w.p.SelectorMultiThread - 收到客户端数据1234567890abcdef 23:06:07 [INFO ] [Boss] c.w.p.SelectorMultiThread - 客户端已连接java.nio.channels.SocketChannel[connected local/127.0.0.1:9527 remote/127.0.0.1:64728] 23:06:07 [INFO ] [Boss] c.w.p.SelectorMultiThread - 客户端channel注册到worker线程的 selector-开始第二版在第一版中【客户端channel注册到worker线程的selector上】的逻辑运行在主线程中【调用worker的selector的select方法】的逻辑运行在worker线程中如果想办法让这两段逻辑运行在同一个线程中就可以保证执行的先后顺序。主线程publicstaticvoidmain(String[]args)throwsIOException{Thread.currentThread().setName(Boss);ServerSocketChannelserverSocketChannelServerSocketChannel.open();serverSocketChannel.bind(newInetSocketAddress(9527));serverSocketChannel.configureBlocking(false);SelectorselectorSelector.open();SelectionKeyselectionKeyserverSocketChannel.register(selector,0,null);selectionKey.interestOps(SelectionKey.OP_ACCEPT);// 创建固定数量的workerWorkerworkernewWorker(worker-0);while(true){selector.select();IteratorSelectionKeyiteratorselector.selectedKeys().iterator();while(iterator.hasNext()){SelectionKeykeyiterator.next();iterator.remove();// 主线程只负责处理 accept事件if(key.isAcceptable()){SocketChannelsocketChannelserverSocketChannel.accept();socketChannel.configureBlocking(false);log.info(客户端已连接socketChannel);// 初始化workerlog.info(客户端channel注册到worker线程的 selector-开始);// 连接建立后将channel注册到worker线程的 selector上, 并且监听read事件worker.init(socketChannel);log.info(客户端channel注册到worker线程的 selector-结束);}}}}worker线程staticclassWorkerimplementsRunnable{privateThreadthread;privateSelectorselector;privateStringname;privatevolatilebooleanstartfalse;privateConcurrentLinkedQueueRunnablequeuenewConcurrentLinkedQueue();publicWorker(Stringname){this.namename;}// 初始化【线程】和【selector】publicvoidinit(SocketChannelsocketChannel)throwsIOException{if(!start){this.threadnewThread(this,name);selectorSelector.open();thread.start();starttrue;}// 向队列中添加任务但是任务没有立刻执行queue.add(()-{try{socketChannel.register(selector,SelectionKey.OP_READ,null);}catch(ClosedChannelExceptione){thrownewRuntimeException(e);}});// 将 selector 从阻塞中唤醒selector.wakeup();}Overridepublicvoidrun(){while(true){try{selector.select();Runnablerunnablequeue.poll();if(runnable!null){runnable.run();}IteratorSelectionKeyiteratorselector.selectedKeys().iterator();while(iterator.hasNext()){SelectionKeykeyiterator.next();iterator.remove();if(key.isReadable()){ByteBufferbyteBufferByteBuffer.allocate(16);SocketChannelchannel(SocketChannel)key.channel();channel.read(byteBuffer);byteBuffer.flip();CharBufferdecodeCharset.defaultCharset().decode(byteBuffer);log.info(String.valueOf(decode));byteBuffer.clear();}}}catch(IOExceptione){thrownewRuntimeException(e);}}}}启动多个客户端可以看到正常连接并且正常接收到数据23:23:52 [INFO ] [Boss] c.w.p.SelectorMultiThread - 客户端已连接java.nio.channels.SocketChannel[connected local/127.0.0.1:9527 remote/127.0.0.1:65125] 23:23:52 [INFO ] [Boss] c.w.p.SelectorMultiThread - 客户端channel注册到worker线程的 selector-开始 23:23:52 [INFO ] [Boss] c.w.p.SelectorMultiThread - 客户端channel注册到worker线程的 selector-结束 23:23:52 [INFO ] [worker-0] c.w.p.SelectorMultiThread - 1234567890abcdef 23:24:03 [INFO ] [Boss] c.w.p.SelectorMultiThread - 客户端已连接java.nio.channels.SocketChannel[connected local/127.0.0.1:9527 remote/127.0.0.1:65137] 23:24:03 [INFO ] [Boss] c.w.p.SelectorMultiThread - 客户端channel注册到worker线程的 selector-开始 23:24:03 [INFO ] [Boss] c.w.p.SelectorMultiThread - 客户端channel注册到worker线程的 selector-结束 23:24:03 [INFO ] [worker-0] c.w.p.SelectorMultiThread - 1234567890abcdef关键点使用队列将【客户端channel注册到selector】的逻辑先保存起来每次有客户端连接都要添加这段逻辑在没有读写事件发生时worker的selector一定是阻塞的下次再有客户端连接一定会注册失败因此要通过selector的wakewp方法唤醒selector除了事件可以唤醒selector也可以手动调用wakeupselector被唤醒后如果队列中有任务说明有客户端channel需要注册客户端channel注册完成后再处理读写事件第三版服务器端创建多个 worker数量为CPU核心数Runtime.getRuntime().availableProcessors()这个方法有坑如果在Docker环境拿到的是物理机的核心数不是容器的核心数。publicstaticvoidmain(String[]args)throwsIOException{Thread.currentThread().setName(Boss);ServerSocketChannelserverSocketChannelServerSocketChannel.open();serverSocketChannel.bind(newInetSocketAddress(9527));serverSocketChannel.configureBlocking(false);SelectorselectorSelector.open();SelectionKeyselectionKeyserverSocketChannel.register(selector,0,null);selectionKey.interestOps(SelectionKey.OP_ACCEPT);// 创建固定数量的workerWorker[]workersnewWorker[Runtime.getRuntime().availableProcessors()];for(inti0;iworkers.length;i){workers[i]newWorker(worker-i);}AtomicIntegercountnewAtomicInteger();while(true){selector.select();IteratorSelectionKeyiteratorselector.selectedKeys().iterator();while(iterator.hasNext()){SelectionKeykeyiterator.next();iterator.remove();// 主线程只负责处理 accept事件if(key.isAcceptable()){SocketChannelsocketChannelserverSocketChannel.accept();socketChannel.configureBlocking(false);log.info(客户端已连接socketChannel);// 初始化workerlog.info(客户端channel注册到worker线程的 selector-开始);// 连接建立后将channel注册到worker线程的 selector上, 并且监听read事件workers[count.getAndIncrement()%workers.length].init(socketChannel);log.info(客户端channel注册到worker线程的 selector-结束);}}}}workerstaticclassWorkerimplementsRunnable{privateThreadthread;privateSelectorselector;privateStringname;privatevolatilebooleanstartfalse;privateConcurrentLinkedQueueRunnablequeuenewConcurrentLinkedQueue();publicWorker(Stringname){this.namename;}// 初始化【线程】和【selector】publicvoidinit(SocketChannelsocketChannel)throwsIOException{if(!start){this.threadnewThread(this,name);selectorSelector.open();thread.start();starttrue;}// 向队列中添加任务但是任务没有立刻执行queue.add(()-{try{socketChannel.register(selector,SelectionKey.OP_READ,null);}catch(ClosedChannelExceptione){thrownewRuntimeException(e);}});// 将 selector 从阻塞中唤醒selector.wakeup();}Overridepublicvoidrun(){while(true){try{selector.select();Runnablerunnablequeue.poll();if(runnable!null){runnable.run();}IteratorSelectionKeyiteratorselector.selectedKeys().iterator();while(iterator.hasNext()){SelectionKeykeyiterator.next();iterator.remove();if(key.isReadable()){ByteBufferbyteBufferByteBuffer.allocate(16);SocketChannelchannel(SocketChannel)key.channel();channel.read(byteBuffer);byteBuffer.flip();CharBufferdecodeCharset.defaultCharset().decode(byteBuffer);log.info(String.valueOf(decode));byteBuffer.clear();}}}catch(IOExceptione){thrownewRuntimeException(e);}}}}