單線程會浪費多核的優勢

單線程如果在某一業務上花費時間過長，會影響其他業務的處理

boss負責連接，worker負責讀寫

服務端代碼：

package com.netty.demo;import java.io.IOException; import java.net.InetSocketAddress; import java.nio.ByteBuffer; import java.nio.channels.SelectionKey; import java.nio.channels.Selector; import java.nio.channels.ServerSocketChannel; import java.nio.channels.SocketChannel; import java.util.Iterator; import java.util.Set; import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.atomic.AtomicInteger;import static com.netty.demo.ByteBufferUtil.debugAll;public class ThreadsServer {public static void main(String[] args) {try (ServerSocketChannel server = ServerSocketChannel.open()) {// 當前線程為Boss線程Thread.currentThread().setName("Boss");server.bind(new InetSocketAddress(8080));// 負責輪詢Accept事件的SelectorSelector boss = Selector.open();server.configureBlocking(false);server.register(boss, SelectionKey.OP_ACCEPT);// 創建固定數量的WorkerWorker[] workers = new Worker[4];// 用于負載均衡的原子整數AtomicInteger robin = new AtomicInteger(0);for(int i = 0; i < workers.length; i++) {workers[i] = new Worker("worker-"+i);}while (true) {boss.select();Set<SelectionKey> selectionKeys = boss.selectedKeys();Iterator<SelectionKey> iterator = selectionKeys.iterator();while (iterator.hasNext()) {SelectionKey key = iterator.next();iterator.remove();// BossSelector負責Accept事件if (key.isAcceptable()) {// 建立連接SocketChannel socket = server.accept();System.out.println("connected...");socket.configureBlocking(false);// socket注冊到Worker的Selector中System.out.println("before read...");// 負載均衡，輪詢分配Workerworkers[robin.getAndIncrement()% workers.length].register(socket);System.out.println("after read...");}}}} catch (IOException e) {e.printStackTrace();}}static class Worker implements Runnable {private Thread thread;private volatile Selector selector;private String name;private volatile boolean started = false;/*** 同步隊列，用于Boss線程與Worker線程之間的通信*/private ConcurrentLinkedQueue<Runnable> queue;public Worker(String name) {this.name = name;}public void register(final SocketChannel socket) throws IOException {// 只啟動一次if (!started) {thread = new Thread(this, name);selector = Selector.open();queue = new ConcurrentLinkedQueue<>();thread.start();started = true;}// 向同步隊列中添加SocketChannel的注冊事件// 在Worker線程中執行注冊事件queue.add(new Runnable() {@Overridepublic void run() {try {socket.register(selector, SelectionKey.OP_READ);} catch (IOException e) {e.printStackTrace();}}});// 喚醒被阻塞的Selector// select類似LockSupport中的park，wakeup的原理類似LockSupport中的unparkselector.wakeup();}@Overridepublic void run() {while (true) {try {selector.select();// 通過同步隊列獲得任務并運行Runnable task = queue.poll();if (task != null) {// 獲得任務，執行注冊操作task.run();}Set<SelectionKey> selectionKeys = selector.selectedKeys();Iterator<SelectionKey> iterator = selectionKeys.iterator();while(iterator.hasNext()) {SelectionKey key = iterator.next();iterator.remove();// Worker只負責Read事件if (key.isReadable()) {// 簡化處理，省略細節SocketChannel socket = (SocketChannel) key.channel();ByteBuffer buffer = ByteBuffer.allocate(16);socket.read(buffer);buffer.flip();ByteBufferUtil.debugAll(buffer);}}} catch (IOException e) {e.printStackTrace();}}}} }

客戶端代碼：

package com.netty.demo;import java.io.IOException; import java.net.InetSocketAddress; import java.nio.channels.SocketChannel; import java.nio.charset.Charset;public class TestClient {public static void main(String[] args) throws IOException{SocketChannel sc = SocketChannel.open();sc.connect(new InetSocketAddress("localhost",8080));sc.write(Charset.defaultCharset().encode("1234567890abcdef"));System.in.read();} }

一般worker的數量設置為cpu的核心數比較合適

阻塞I/O模型

非阻塞I/O模型：

多路復用I/O模型：

零拷貝：

?

總結

以上是生活随笔為你收集整理的NIO的多线程优化的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。