折騰了我一周，原來Netty網路編程就是這麼個破玩意兒！！！

来源:https://www.cnblogs.com/jiagooushi/archive/2022/12/29/17012450.html
-Advertisement-
Play Games

1、阻塞 阻塞模式下，相關方法都會導致線程暫停 ServerSocketChannel.accept 會在沒有連接建立時讓線程暫停 SocketChannel.read 會在通道中沒有數據可讀時讓線程暫停 阻塞的表現其實就是線程暫停了，暫停期間不會占用 cpu，但線程相當於閑置 單線程下，阻塞方法之 ...


1、阻塞

  • 阻塞模式下，相關方法都會導致線程暫停
    • ServerSocketChannel.accept 會在沒有連接建立時讓線程暫停
    • SocketChannel.read 會在通道中沒有數據可讀時讓線程暫停
    • 阻塞的表現其實就是線程暫停了，暫停期間不會占用 cpu，但線程相當於閑置
  • 單線程下，阻塞方法之間相互影響，幾乎不能正常工作，需要多線程支持
  • 但多線程下，有新的問題，體現在以下方面
    • 32 位 jvm 一個線程 320k，64 位 jvm 一個線程 1024k，如果連接數過多，必然導致 OOM，並且線程太多，反而會因為頻繁上下文切換導致性能降低
    • 可以採用線程池技術來減少線程數和線程上下文切換，但治標不治本，如果有很多連接建立，但長時間 inactive，會阻塞線程池中所有線程，因此不適合長連接，只適合短連接

服務端代碼

public class Server {
    public static void main(String[] args) {
        // 創建緩衝區
        ByteBuffer buffer = ByteBuffer.allocate(16);
        // 獲得伺服器通道
        try(ServerSocketChannel server = ServerSocketChannel.open()) {
            // 為伺服器通道綁定埠
            server.bind(new InetSocketAddress(8080));
            // 用戶存放連接的集合
            ArrayList<SocketChannel> channels = new ArrayList<>();
            // 迴圈接收連接
            while (true) {
                System.out.println("before connecting...");
                // 沒有連接時，會阻塞線程
                SocketChannel socketChannel = server.accept();
                System.out.println("after connecting...");
                channels.add(socketChannel);
                // 迴圈遍歷集合中的連接
                for(SocketChannel channel : channels) {
                    System.out.println("before reading");
                    // 處理通道中的數據
                    // 當通道中沒有數據可讀時，會阻塞線程
                    channel.read(buffer);
                    buffer.flip();
                    ByteBufferUtil.debugRead(buffer);
                    buffer.clear();
                    System.out.println("after reading");
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

客戶端代碼

public class Client {
    public static void main(String[] args) {
        try (SocketChannel socketChannel = SocketChannel.open()) {
            // 建立連接
            socketChannel.connect(new InetSocketAddress("localhost", 8080));
            System.out.println("waiting...");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

運行結果

  • 客戶端 - 伺服器建立連接前：伺服器端因 accept 阻塞

file

  • 客戶端 - 伺服器建立連接後，客戶端發送消息前：伺服器端因通道為空被阻塞

file

  • 客戶端發送數據後，伺服器處理通道中的數據。再次進入迴圈時，再次被 accept 阻塞

file

  • 之前的客戶端再次發送消息，伺服器端因為被 accept 阻塞，無法處理之前客戶端發送到通道中的信息

file

2、非阻塞

  • 可以通過 ServerSocketChannel 的 configureBlocking (false) 方法將 獲得連接設置為非阻塞的。此時若沒有連接，accept 會返回 null
  • 可以通過 SocketChannel 的 configureBlocking (false) 方法將從通道中 讀取數據設置為非阻塞的。若此時通道中沒有數據可讀，read 會返回 - 1

伺服器代碼如下

public class Server {
    public static void main(String[] args) {
        // 創建緩衝區
        ByteBuffer buffer = ByteBuffer.allocate(16);
        // 獲得伺服器通道
        try(ServerSocketChannel server = ServerSocketChannel.open()) {
            // 設置為非阻塞模式，沒有連接時返回null，不會阻塞線程
            server.configureBlocking(false);
            // 為伺服器通道綁定埠
            server.bind(new InetSocketAddress(8080));
            // 用戶存放連接的集合
            ArrayList<SocketChannel> channels = new ArrayList<>();
            // 迴圈接收連接
            while (true) {
                
              
                SocketChannel socketChannel = server.accept();
                // 通道不為空時才將連接放入到集合中
                if (socketChannel != null) {
                    System.out.println("after connecting...");
                    channels.add(socketChannel);
                }
                // 迴圈遍歷集合中的連接
                for(SocketChannel channel : channels) {
                    // 處理通道中的數據
                    // 設置為非阻塞模式，若通道中沒有數據，會返回0，不會阻塞線程
                    channel.configureBlocking(false);
                    int read = channel.read(buffer);
                    if(read > 0) {
                        buffer.flip();
                        ByteBufferUtil.debugRead(buffer);
                        buffer.clear();
                        System.out.println("after reading");
                    }
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

這樣寫存在一個問題，因為設置為了非阻塞，會一直執行 while (true) 中的代碼，CPU 一直處於忙碌狀態，會使得性能變低，所以實際情況中不使用這種方法處理請求

3、Selector

多路復用

單線程可以配合 Selector 完成對多個 Channel 可讀寫事件的監控，這稱之為多路復用

  • 多路復用僅針對網路 IO，普通文件 IO 無法利用多路復用
  • 如果不用 Selector 的非阻塞模式，線程大部分時間都在做無用功，而 Selector 能夠保證
    • 有可連接事件時才去連接
    • 有可讀事件才去讀取
    • 有可寫事件才去寫入
      • 限於網路傳輸能力，Channel 未必時時可寫，一旦 Channel 可寫，會觸發 Selector 的可寫事件

4、使用及 Accpet 事件

要使用 Selector 實現多路復用，服務端代碼如下改進

public class SelectServer {
    public static void main(String[] args) {
        ByteBuffer buffer = ByteBuffer.allocate(16);
        // 獲得伺服器通道
        try(ServerSocketChannel server = ServerSocketChannel.open()) {
            server.bind(new InetSocketAddress(8080));
            // 創建選擇器
            Selector selector = Selector.open();
            
            // 通道必須設置為非阻塞模式
            server.configureBlocking(false);
            // 將通道註冊到選擇器中，並設置感興趣的事件
            server.register(selector, SelectionKey.OP_ACCEPT);
            while (true) {
                // 若沒有事件就緒，線程會被阻塞，反之不會被阻塞。從而避免了CPU空轉
                // 返回值為就緒的事件個數
                int ready = selector.select();
                System.out.println("selector ready counts : " + ready);
                
                // 獲取所有事件
                Set<SelectionKey> selectionKeys = selector.selectedKeys();
                
                // 使用迭代器遍歷事件
                Iterator<SelectionKey> iterator = selectionKeys.iterator();
                while (iterator.hasNext()) {
                    SelectionKey key = iterator.next();
                    
                    // 判斷key的類型
                    if(key.isAcceptable()) {
                        // 獲得key對應的channel
                        ServerSocketChannel channel = (ServerSocketChannel) key.channel();
                        System.out.println("before accepting...");
                        
        				// 獲取連接並處理，而且是必須處理，否則需要取消
                        SocketChannel socketChannel = channel.accept();
                        System.out.println("after accepting...");
                        
                        // 處理完畢後移除
                        iterator.remove();
                    }
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

步驟解析

  • 獲得選擇器 Selector
Selector selector = Selector.open();
  • 將通道設置為非阻塞模式，並註冊到選擇器中，並設置感興趣的事件
    • channel 必須工作在非阻塞模式
    • FileChannel 沒有非阻塞模式，因此不能配合 selector 一起使用
    • 綁定的事件類型可以有
      • connect - 客戶端連接成功時觸發
  • accept - 伺服器端成功接受連接時觸發
    • read - 數據可讀入時觸發，有因為接收能力弱，數據暫不能讀入的情況
  • write - 數據可寫出時觸發，有因為發送能力弱，數據暫不能寫出的情況
// 通道必須設置為非阻塞模式
server.configureBlocking(false);
// 將通道註冊到選擇器中，並設置感興趣的實踐
server.register(selector, SelectionKey.OP_ACCEPT);

  • 通過 Selector 監聽事件，並獲得就緒的通道個數，若沒有通道就緒，線程會被阻塞

    • 阻塞直到綁定事件發生

      int count = selector.select();

    • 阻塞直到綁定事件發生，或是超時（時間單位為 ms）

      int count = selector.select(long timeout);

    • 不會阻塞，也就是不管有沒有事件，立刻返回，自己根據返回值檢查是否有事件

      int count = selector.selectNow();

  • 獲取就緒事件並得到對應的通道，然後進行處理

// 獲取所有事件
Set<SelectionKey> selectionKeys = selector.selectedKeys();
                
// 使用迭代器遍歷事件
Iterator<SelectionKey> iterator = selectionKeys.iterator();

while (iterator.hasNext()) {
	SelectionKey key = iterator.next();
                    
	// 判斷key的類型，此處為Accept類型
	if(key.isAcceptable()) {
        // 獲得key對應的channel
        ServerSocketChannel channel = (ServerSocketChannel) key.channel();

        // 獲取連接並處理，而且是必須處理，否則需要取消
        SocketChannel socketChannel = channel.accept();

        // 處理完畢後移除
        iterator.remove();
	}
}

事件發生後能否不處理

事件發生後，要麼處理，要麼取消（cancel），不能什麼都不做，否則下次該事件仍會觸發，這是因為 nio 底層使用的是水平觸發

5、Read 事件

  • 在 Accept 事件中，若有客戶端與伺服器端建立了連接，需要將其對應的 SocketChannel 設置為非阻塞，並註冊到選擇其中
    添加 Read 事件，觸發後進行讀取操作
  • 添加 Read 事件，觸發後進行讀取操作
public class SelectServer {
    public static void main(String[] args) {
        ByteBuffer buffer = ByteBuffer.allocate(16);
        // 獲得伺服器通道
        try(ServerSocketChannel server = ServerSocketChannel.open()) {
            server.bind(new InetSocketAddress(8080));
            // 創建選擇器
            Selector selector = Selector.open();
            // 通道必須設置為非阻塞模式
            server.configureBlocking(false);
            // 將通道註冊到選擇器中，並設置感興趣的實踐
            server.register(selector, SelectionKey.OP_ACCEPT);
            // 為serverKey設置感興趣的事件
            while (true) {
                // 若沒有事件就緒，線程會被阻塞，反之不會被阻塞。從而避免了CPU空轉
                // 返回值為就緒的事件個數
                int ready = selector.select();
                System.out.println("selector ready counts : " + ready);
                // 獲取所有事件
                Set<SelectionKey> selectionKeys = selector.selectedKeys();
                // 使用迭代器遍歷事件
                Iterator<SelectionKey> iterator = selectionKeys.iterator();
                while (iterator.hasNext()) {
                    SelectionKey key = iterator.next();
                    // 判斷key的類型
                    if(key.isAcceptable()) {
                        // 獲得key對應的channel
                        ServerSocketChannel channel = (ServerSocketChannel) key.channel();
                        System.out.println("before accepting...");
                        // 獲取連接
                        SocketChannel socketChannel = channel.accept();
                        System.out.println("after accepting...");
                        // 設置為非阻塞模式，同時將連接的通道也註冊到選擇其中
                        socketChannel.configureBlocking(false);
                        socketChannel.register(selector, SelectionKey.OP_READ);
                        // 處理完畢後移除
                        iterator.remove();
                    } else if (key.isReadable()) {
                        SocketChannel channel = (SocketChannel) key.channel();
                        System.out.println("before reading...");
                        channel.read(buffer);
                        System.out.println("after reading...");
                        buffer.flip();
                        ByteBufferUtil.debugRead(buffer);
                        buffer.clear();
                        // 處理完畢後移除
                        iterator.remove();
                    }
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

刪除事件

當處理完一個事件後，一定要調用迭代器的 remove 方法移除對應事件，否則會出現錯誤。原因如下

以我們上面的 Read 事件 的代碼為例

  • 當調用了 server.register (selector, SelectionKey.OP_ACCEPT) 後，Selector 中維護了一個集合，用於存放 SelectionKey 以及其對應的通道

    // WindowsSelectorImpl 中的 SelectionKeyImpl數組
private SelectionKeyImpl[] channelArray = new SelectionKeyImpl[8];

    public class SelectionKeyImpl extends AbstractSelectionKey {
    // Key對應的通道
    final SelChImpl channel;
    ...
}

file

  • 當選擇器中的通道對應的事件發生後，selecionKey 會被放到另一個集合中，但是 selecionKey 不會自動移除，所以需要我們在處理完一個事件後，通過迭代器手動移除其中的 selecionKey。否則會導致已被處理過的事件再次被處理，就會引發錯誤

file

斷開處理

當客戶端與伺服器之間的連接斷開時，會給伺服器端發送一個讀事件，對異常斷開和正常斷開需要加以不同的方式進行處理

  • 正常斷開

    • 正常斷開時，伺服器端的 channel.read (buffer) 方法的返回值為 - 1，所以當結束到返回值為 - 1 時，需要調用 key 的 cancel 方法取消此事件，併在取消後移除該事件

      int read = channel.read(buffer);
// 斷開連接時，客戶端會向伺服器發送一個寫事件，此時read的返回值為-1
if(read == -1) {
    // 取消該事件的處理
	key.cancel();
    channel.close();
} else {
    ...
}
// 取消或者處理，都需要移除key
iterator.remove();

  • 異常斷開

    • 異常斷開時，會拋出 IOException 異常， 在 try-catch 的 catch 塊中捕獲異常並調用 key 的 cancel 方法即可

消息邊界

不處理消息邊界存在的問題

將緩衝區的大小設置為 4 個位元組，發送 2 個漢字（你好），通過 decode 解碼並列印時，會出現亂碼

ByteBuffer buffer = ByteBuffer.allocate(4);
// 解碼並列印
System.out.println(StandardCharsets.UTF_8.decode(buffer));
你�
��

這是因為 UTF-8 字元集下，1 個漢字占用 3 個位元組，此時緩衝區大小為 4 個位元組，一次讀時間無法處理完通道中的所有數據，所以一共會觸發兩次讀事件。這就導致 你好 的 好 字被拆分為了前半部分和後半部分發送，解碼時就會出現問題

處理消息邊界

傳輸的文本可能有以下三種情況

  • 文本大於緩衝區大小
    • 此時需要將緩衝區進行擴容
  • 發生半包現象
  • 發生粘包現象

file

解決思路大致有以下三種

  • 固定消息長度，數據包大小一樣，伺服器按預定長度讀取，當發送的數據較少時，需要將數據進行填充，直到長度與消息規定長度一致。缺點是浪費帶寬

  • 另一種思路是按分隔符拆分，缺點是效率低，需要一個一個字元地去匹配分隔符

  • TLV 格式，即 Type 類型、Length 長度、Value 數據

    （也就是在消息開頭用一些空間存放後面數據的長度），如 HTTP 請求頭中的 Content-Type 與 Content-Length

    。類型和長度已知的情況下，就可以方便獲取消息大小，分配合適的 buffer，缺點是 buffer 需要提前分配，如果內容過大，則影響 server 吞吐量

    • Http 1.1 是 TLV 格式

  • Http 2.0 是 LTV 格式

file

下文的消息邊界處理方式為第二種：按分隔符拆分

附件與擴容

Channel 的 register 方法還有第三個參數：附件，可以向其中放入一個 Object 類型的對象，該對象會與登記的 Channel 以及其對應的 SelectionKey 綁定，可以從 SelectionKey 獲取到對應通道的附件

public final SelectionKey register(Selector sel, int ops, Object att)

可通過 SelectionKey 的 attachment () 方法獲得附件

ByteBuffer buffer = (ByteBuffer) key.attachment();

我們需要在 Accept 事件發生後，將通道註冊到 Selector 中時，對每個通道添加一個 ByteBuffer 附件，讓每個通道發生讀事件時都使用自己的通道，避免與其他通道發生衝突而導致問題

// 設置為非阻塞模式，同時將連接的通道也註冊到選擇其中，同時設置附件
socketChannel.configureBlocking(false);
ByteBuffer buffer = ByteBuffer.allocate(16);
// 添加通道對應的Buffer附件
socketChannel.register(selector, SelectionKey.OP_READ, buffer);

當 Channel 中的數據大於緩衝區時，需要對緩衝區進行擴容操作。此代碼中的擴容的判定方法： Channel 調用 compact 方法後，的 position 與 limit 相等，說明緩衝區中的數據並未被讀取（容量太小），此時創建新的緩衝區，其大小擴大為兩倍。同時還要將舊緩衝區中的數據拷貝到新的緩衝區中，同時調用 SelectionKey 的 attach 方法將新的緩衝區作為新的附件放入 SelectionKey 中

// 如果緩衝區太小，就進行擴容
if (buffer.position() == buffer.limit()) {
    ByteBuffer newBuffer = ByteBuffer.allocate(buffer.capacity()*2);
    // 將舊buffer中的內容放入新的buffer中
    ewBuffer.put(buffer);
    // 將新buffer作為附件放到key中
    key.attach(newBuffer);
}

改造後的伺服器代碼如下

public class SelectServer {
    public static void main(String[] args) {
        // 獲得伺服器通道
        try(ServerSocketChannel server = ServerSocketChannel.open()) {
            server.bind(new InetSocketAddress(8080));
            // 創建選擇器
            Selector selector = Selector.open();
            // 通道必須設置為非阻塞模式
            server.configureBlocking(false);
            // 將通道註冊到選擇器中，並設置感興趣的事件
            server.register(selector, SelectionKey.OP_ACCEPT);
            // 為serverKey設置感興趣的事件
            while (true) {
                // 若沒有事件就緒，線程會被阻塞，反之不會被阻塞。從而避免了CPU空轉
                // 返回值為就緒的事件個數
                int ready = selector.select();
                System.out.println("selector ready counts : " + ready);
                // 獲取所有事件
                Set<SelectionKey> selectionKeys = selector.selectedKeys();
                // 使用迭代器遍歷事件
                Iterator<SelectionKey> iterator = selectionKeys.iterator();
                
                while (iterator.hasNext()) {
                    SelectionKey key = iterator.next();
                    iterator.remove();
                    // 判斷key的類型
                    if(key.isAcceptable()) {
                        // 獲得key對應的channel
                        ServerSocketChannel channel = (ServerSocketChannel) key.channel();
                        System.out.println("before accepting...");
                        // 獲取連接
                        SocketChannel socketChannel = channel.accept();
                        System.out.println("after accepting...");
                        // 設置為非阻塞模式，同時將連接的通道也註冊到選擇其中，同時設置附件
                        socketChannel.configureBlocking(false);
                        ByteBuffer buffer = ByteBuffer.allocate(16);
                        socketChannel.register(selector, SelectionKey.OP_READ, buffer);
                        
                        
                    } else if (key.isReadable()) {
                        SocketChannel channel = (SocketChannel) key.channel();
                        System.out.println("before reading...");
                        // 通過key獲得附件（buffer）
                        ByteBuffer buffer = (ByteBuffer) key.attachment();
                        int read = channel.read(buffer);
                        if(read == -1) {
                            key.cancel();
                            channel.close();
                        } else {
                            // 通過分隔符來分隔buffer中的數據
                            split(buffer);
                            // 如果緩衝區太小，就進行擴容
                            if (buffer.position() == buffer.limit()) {
                                ByteBuffer newBuffer = ByteBuffer.allocate(buffer.capacity()*2);
                                // 將舊buffer中的內容放入新的buffer中
                                buffer.flip();
                                newBuffer.put(buffer);
                                // 將新buffer放到key中作為附件
                                key.attach(newBuffer);
                            }
                        }
                        System.out.println("after reading...");
                       
                       
                    }
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    private static void split(ByteBuffer buffer) {
        buffer.flip();
        for(int i = 0; i < buffer.limit(); i++) {
            // 遍歷尋找分隔符
            // get(i)不會移動position
            if (buffer.get(i) == '\n') {
                // 緩衝區長度
                int length = i+1-buffer.position();
                ByteBuffer target = ByteBuffer.allocate(length);
                // 將前面的內容寫入target緩衝區
                for(int j = 0; j < length; j++) {
                    // 將buffer中的數據寫入target中
                    target.put(buffer.get());
                }
                // 列印結果
                ByteBufferUtil.debugAll(target);
            }
        }
        // 切換為寫模式，但是緩衝區可能未讀完，這裡需要使用compact
        buffer.compact();
    }
}

ByteBuffer 的大小分配

  • 每個 channel 都需要記錄可能被切分的消息，因為 ByteBuffer 不能被多個 channel 共同使用，因此需要為每個 channel 維護一個獨立的 ByteBuffer
  • ByteBuffer 不能太大，比如一個 ByteBuffer 1Mb 的話，要支持百萬連接就要 1Tb 記憶體，因此需要設計大小可變的 ByteBuffer
  • 分配思路可以參考
    • 一種思路是首先分配一個較小的 buffer，例如 4k，如果發現數據不夠，再分配 8k 的 buffer，將 4k buffer 內容拷貝至 8k buffer，優點是消息連續容易處理，缺點是數據拷貝耗費性能
    • 另一種思路是用多個數組組成 buffer，一個數組不夠，把多出來的內容寫入新的數組，與前面的區別是消息存儲不連續解析複雜，優點是避免了拷貝引起的性能損耗

6、Write 事件

伺服器通過 Buffer 向通道中寫入數據時，可能因為通道容量小於 Buffer 中的數據大小，導致無法一次性將 Buffer 中的數據全部寫入到 Channel 中，這時便需要分多次寫入，具體步驟如下

  • 執行一次寫操作，向將 buffer 中的內容寫入到 SocketChannel 中，然後判斷 Buffer 中是否還有數據

  • 若 Buffer 中還有數據，則需要將 SockerChannel 註冊到 Seletor 中，並關註寫事件，同時將未寫完的 Buffer 作為附件一起放入到 SelectionKey 中

 int write = socket.write(buffer);
// 通道中可能無法放入緩衝區中的所有數據
if (buffer.hasRemaining()) {
    // 註冊到Selector中，關註可寫事件，並將buffer添加到key的附件中
    socket.configureBlocking(false);
    socket.register(selector, SelectionKey.OP_WRITE, buffer);
}
  • 添加寫事件的相關操作 key.isWritable()，對 Buffer 再次進行寫操作
    • 每次寫後需要判斷 Buffer 中是否還有數據（是否寫完）。若寫完，需要移除 SelecionKey 中的 Buffer 附件，避免其占用過多記憶體，同時還需移除對寫事件的關註
SocketChannel socket = (SocketChannel) key.channel();
// 獲得buffer
ByteBuffer buffer = (ByteBuffer) key.attachment();
// 執行寫操作
int write = socket.write(buffer);
System.out.println(write);
// 如果已經完成了寫操作，需要移除key中的附件，同時不再對寫事件感興趣
if (!buffer.hasRemaining()) {
    key.attach(null);
    key.interestOps(0);
}

整體代碼如下

public class WriteServer {
    public static void main(String[] args) {
        try(ServerSocketChannel server = ServerSocketChannel.open()) {
            server.bind(new InetSocketAddress(8080));
            server.configureBlocking(false);
            Selector selector = Selector.open();
            server.register(selector, SelectionKey.OP_ACCEPT);
            while (true) {
                selector.select();
                Set<SelectionKey> selectionKeys = selector.selectedKeys();
                Iterator<SelectionKey> iterator = selectionKeys.iterator();
                while (iterator.hasNext()) {
                    SelectionKey key = iterator.next();
                    // 處理後就移除事件
                    iterator.remove();
                    if (key.isAcceptable()) {
                        // 獲得客戶端的通道
                        SocketChannel socket = server.accept();
                        // 寫入數據
                        StringBuilder builder = new StringBuilder();
                        for(int i = 0; i < 500000000; i++) {
                            builder.append("a");
                        }
                        ByteBuffer buffer = StandardCharsets.UTF_8.encode(builder.toString());
                        // 先執行一次Buffer->Channel的寫入，如果未寫完，就添加一個可寫事件
                        int write = socket.write(buffer);
                        System.out.println(write);
                        // 通道中可能無法放入緩衝區中的所有數據
                        if (buffer.hasRemaining()) {
                            // 註冊到Selector中，關註可寫事件，並將buffer添加到key的附件中
                            socket.configureBlocking(false);
                            socket.register(selector, SelectionKey.OP_WRITE, buffer);
                        }
                    } else if (key.isWritable()) {
                        SocketChannel socket = (SocketChannel) key.channel();
                        // 獲得buffer
                        ByteBuffer buffer = (ByteBuffer) key.attachment();
                        // 執行寫操作
                        int write = socket.write(buffer);
                        System.out.println(write);
                        // 如果已經完成了寫操作，需要移除key中的附件，同時不再對寫事件感興趣
                        if (!buffer.hasRemaining()) {
                            key.attach(null);
                            key.interestOps(0);
                        }
                    }
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

7、優化

多線程優化

充分利用多核 CPU，分兩組選擇器

  • 單線程配一個選擇器（Boss），專門處理 accept 事件
  • 創建 cpu 核心數的線程（Worker），每個線程配一個選擇器，輪流處理 read 事件
實現思路

  • 創建一個負責處理 Accept 事件的 Boss 線程，與多個負責處理 Read 事件的 Worker 線程

  • Boss 線程執行的操作

    • 接受並處理 Accepet 事件，當 Accept 事件發生後，調用 Worker 的 register (SocketChannel socket) 方法，讓 Worker 去處理 Read 事件，其中需要根據標識 robin 去判斷將任務分配給哪個 Worker

      // 創建固定數量的Worker
Worker[] workers = new Worker[4];
// 用於負載均衡的原子整數
AtomicInteger robin = new AtomicInteger(0);
// 負載均衡，輪詢分配Worker
workers[robin.getAndIncrement()% workers.length].register(socket);

    • register (SocketChannel socket) 方法會通過同步隊列完成 Boss 線程與 Worker 線程之間的通信，讓 SocketChannel 的註冊任務被 Worker 線程執行。添加任務後需要調用 selector.wakeup () 來喚醒被阻塞的 Selector

      public void register(final SocketChannel socket) throws IOException {
    // 只啟動一次
    if (!started) {
       // 初始化操作
    }
    // 向同步隊列中添加SocketChannel的註冊事件
    // 在Worker線程中執行註冊事件
    queue.add(new Runnable() {
        @Override
        public void run() {
            try {
                socket.register(selector, SelectionKey.OP_READ);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    });
    // 喚醒被阻塞的Selector
    // select類似LockSupport中的park，wakeup的原理類似LockSupport中的unpark
    selector.wakeup();
}

  • Worker 線程執行的操作

    • 從同步隊列中獲取註冊任務，並處理 Read 事件
實現代碼
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事件的Selector
            Selector boss = Selector.open();
            server.configureBlocking(false);
            server.register(boss, SelectionKey.OP_ACCEPT);
            // 創建固定數量的Worker
            Worker[] 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...");
                        // 負載均衡，輪詢分配Worker
                        workers[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() {
                @Override
                public void run() {
                    try {
                        socket.register(selector, SelectionKey.OP_READ);
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }
            });
            // 喚醒被阻塞的Selector
            // select類似LockSupport中的park，wakeup的原理類似LockSupport中的unpark
            selector.wakeup();
        }

        @Override
        public 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();
                }
            }
        }
    }
}

本文由傳智教育博學谷教研團隊發佈。

如果本文對您有幫助，歡迎關註和點贊；如果您有任何建議也可留言評論或私信，您的支持是我堅持創作的動力。

轉載請註明出處！


您的分享是我們最大的動力!

-Advertisement-
Play Games
更多相關文章
  • VueJS使用addEventListener的事件如何觸發執行函數的this
    1、使用瀏覽器監聽切屏為例 此處為考慮瀏覽器相容性推薦使用：document.addEventListener 1.1、正常函數使用如下： let n = 0; let max = 3; // 切屏最大次數 document.addEventListener("visibilitychange", ...
  • TypeScript 前端工程最佳實踐
    作者：王春雨 前言 隨著前端工程化的快速發展， TypeScript 變得越來越受歡迎，它已經成為前端開發人員必備技能。 TypeScript 最初是由微軟開發並開源的一種編程語言，自2012年10月發佈首個公開版本以來，它已得到了人們的廣泛認可。TypeScript 發展至今，已經成為很多大型項目 ...
  • 常用設計模式之簡單工廠模式
    簡單工廠模式 （C++） 工廠模式的使用場景也比較多，像是一些組件對象的創建場景，這裡說一下簡單工廠模式。 特點及應用 1. 特點 （1）封裝具體實現，只提供管理對象介面 舉個例子：我現在需要一個界面工廠，工廠只需要給我對應界面就可以，而我不需要去關心界面裡面是怎樣實現的，如何創建的。 （2）集中管 ...
  • 淺談系統性能提升的經驗和方法
    資金核對的數據組裝-執行-應急鏈路，有著千萬級TPS併發量，同時由於資金業務特性，對系統可用性和準確性要求非常高；日常開發過程中會遇到各種各樣的高可用問題，也在不斷地嘗試做一些系統設計以及性能優化，在此期間總結了部分性能優化的經驗和方法，跟大家一起分享和交流。 ...
  • day11-功能實現10
    家居網購項目實現010 以下皆為部分代碼，詳見 https://github.com/liyuelian/furniture_mall.git 24.bugFix-添加購物車按鈕動態處理 24.1需求分析/圖解 如某個家居的庫存量為0，前臺的“add to cart”按鈕顯示為“暫時缺貨” 後臺也要 ...
  • C++11：for迴圈（基於範圍的迴圈）
    1. C++ 98/03標準的for迴圈 在C++ 98/03標準中，如果要用 for 迴圈語句遍歷一個數組或者容器，只能套用如下結構： for(表達式 1; 表達式 2; 表達式 3){ //迴圈體 } 例如，下麵程式演示了用上述結構遍曆數組和容器的具體實現過程： #include <iostre ...
  • 7個你可能從未使用過的Python隱藏技巧
    Python 是每個程式員都喜歡的語言，因為它易於編碼和易於閱讀的語法。但是，你知道 python 有一些很酷的技巧可以用來讓事情變得更簡單嗎？在今天的內容中，我將與你分享7 個你可能從未使用過的Python 技巧。 1、功能屬性 這種 hack 類似於類和對象概念。現在，可以聲明，也可以稍後在程式 ...
  • 每日演算法之把字元串轉換成整數(atoi)
    JZ67 把字元串轉換成整數(atoi) 題目 寫一個函數 StrToInt，實現把字元串轉換成整數這個功能。不能使用 atoi 或者其他類似的庫函數。傳入的字元串可能有以下部分組成: 1.若幹空格 2.（可選）一個符號字元（'+' 或 '-'） 3. 數字，字母，符號，空格組成的字元串表達式 4. ...
一周排行
    -Advertisement-
    Play Games
  • 移動開發（一）：使用.NET MAUI開發第一個安卓APP
    移動開發（一）：使用.NET MAUI開發第一個安卓APP 對於工作多年的C#程式員來說，近來想嘗試開發一款安卓APP，考慮了很久最終選擇使用.NET MAUI這個微軟官方的框架來嘗試體驗開發安卓APP，畢竟是使用Visual Studio開發工具，使用起來也比較的順手，結合微軟官方的教程進行了安卓 ...
  • wpf ToggleButton選中效果和一個登錄界面
    前言 QuestPDF 是一個開源 .NET 庫，用於生成 PDF 文檔。使用了C# Fluent API方式可簡化開發、減少錯誤並提高工作效率。利用它可以輕鬆生成 PDF 報告、發票、導出文件等。 項目介紹 QuestPDF 是一個革命性的開源 .NET 庫，它徹底改變了我們生成 PDF 文檔的方 ...
  • 彈幕樹洞項目功能新增篇
    項目地址 項目後端地址： https://github.com/ZyPLJ/ZYTteeHole 項目前端頁面地址： ZyPLJ/TreeHoleVue (github.com) https://github.com/ZyPLJ/TreeHoleVue 目前項目測試訪問地址： http://tree ...
  • 第27篇 sqlserver2022詳細安裝步驟
    話不多說，直接開乾 一.下載 1.官方鏈接下載： https://www.microsoft.com/zh-cn/sql-server/sql-server-downloads 2.在下載目錄中找到下麵這個小的安裝包 SQL2022-SSEI-Dev.exe，運行開始下載SQL server； 二. ...
  • .NET 開源高性能 MQTT 類庫
    前言 隨著物聯網（IoT）技術的迅猛發展，MQTT（消息隊列遙測傳輸）協議憑藉其輕量級和高效性，已成為眾多物聯網應用的首選通信標準。 MQTTnet 作為一個高性能的 .NET 開源庫，為 .NET 平臺上的 MQTT 客戶端與伺服器開發提供了強大的支持。 本文將全面介紹 MQTTnet 的核心功能 ...
  • Serilog文檔翻譯系列（六） - 可用的接收器、增強器、格式化輸出
    Serilog支持多種接收器用於日誌存儲，增強器用於添加屬性，LogContext管理動態屬性，支持多種輸出格式包括純文本、JSON及ExpressionTemplate。還提供了自定義格式化選項，適用於不同需求。 ...
  • 警惕 Visual Studio 屬性求值副作用導致邏輯不符合預期
    目錄簡介獲取 HTML 文檔解析 HTML 文檔測試參考文章 簡介 動態內容網站使用 JavaScript 腳本動態檢索和渲染數據，爬取信息時需要模擬瀏覽器行為，否則獲取到的源碼基本是空的。 本文使用的爬取步驟如下： 使用 Selenium 獲取渲染後的 HTML 文檔 使用 HtmlAgility ...
  • [使用目前最新版]HybridCLR6.9.0+YooAsset2.2.4實現純C# Unity熱更新方案 (一)
    1.前言 什麼是熱更新 游戲或者軟體更新時，無需重新下載客戶端進行安裝，而是在應用程式啟動的情況下，在內部進行資源或者代碼更新 Unity目前常用熱更新解決方案 HybridCLR，Xlua，ILRuntime等 Unity目前常用資源管理解決方案 AssetBundles，Addressable， ...
  • 在 ASP.NET Core Web API 中使用操作篩選器統一處理通用操作
    本文章主要是在C# ASP.NET Core Web API框架實現向手機發送驗證碼簡訊功能。這裡我選擇是一個互億無線簡訊驗證碼平臺，其實像阿裡雲，騰訊雲上面也可以。 首先我們先去 互億無線 https://www.ihuyi.com/api/sms.html 去註冊一個賬號 註冊完成賬號後，它會送 ...
  • 第28篇 如何.net中實現高效可靠數據同步api
    通過以下方式可以高效，並保證數據同步的可靠性 1.API設計 使用RESTful設計，確保API端點明確，並使用適當的HTTP方法（如POST用於創建，PUT用於更新）。 設計清晰的請求和響應模型，以確保客戶端能夠理解預期格式。 2.數據驗證 在伺服器端進行嚴格的數據驗證，確保接收到的數據符合預期格 ...
所有分類