單線程會(huì)浪費(fèi)多核的優(yōu)勢(shì)

單線程如果在某一業(yè)務(wù)上花費(fèi)時(shí)間過(guò)長(zhǎng)，會(huì)影響其他業(yè)務(wù)的處理

boss負(fù)責(zé)連接，worker負(fù)責(zé)讀寫

服務(wù)端代碼：

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()) {// 當(dāng)前線程為Boss線程Thread.currentThread().setName("Boss");server.bind(new InetSocketAddress(8080));// 負(fù)責(zé)輪詢Accept事件的SelectorSelector boss = Selector.open();server.configureBlocking(false);server.register(boss, SelectionKey.OP_ACCEPT);// 創(chuàng)建固定數(shù)量的WorkerWorker[] workers = new Worker[4];// 用于負(fù)載均衡的原子整數(shù)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負(fù)責(zé)Accept事件if (key.isAcceptable()) {// 建立連接SocketChannel socket = server.accept();System.out.println("connected...");socket.configureBlocking(false);// socket注冊(cè)到Worker的Selector中System.out.println("before read...");// 負(fù)載均衡，輪詢分配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;/*** 同步隊(duì)列，用于Boss線程與Worker線程之間的通信*/private ConcurrentLinkedQueue<Runnable> queue;public Worker(String name) {this.name = name;}public void register(final SocketChannel socket) throws IOException {// 只啟動(dòng)一次if (!started) {thread = new Thread(this, name);selector = Selector.open();queue = new ConcurrentLinkedQueue<>();thread.start();started = true;}// 向同步隊(duì)列中添加SocketChannel的注冊(cè)事件// 在Worker線程中執(zhí)行注冊(cè)事件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();// 通過(guò)同步隊(duì)列獲得任務(wù)并運(yùn)行Runnable task = queue.poll();if (task != null) {// 獲得任務(wù)，執(zhí)行注冊(cè)操作task.run();}Set<SelectionKey> selectionKeys = selector.selectedKeys();Iterator<SelectionKey> iterator = selectionKeys.iterator();while(iterator.hasNext()) {SelectionKey key = iterator.next();iterator.remove();// Worker只負(fù)責(zé)Read事件if (key.isReadable()) {// 簡(jiǎn)化處理，省略細(xì)節(jié)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的數(shù)量設(shè)置為cpu的核心數(shù)比較合適

阻塞I/O模型

非阻塞I/O模型：

多路復(fù)用I/O模型：

零拷貝：

?

總結(jié)

以上是生活随笔為你收集整理的NIO的多线程优化的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。