您现在的位置：程式碼之家 >> 編程語言 > >> Java 程式設計 >> Java 相關知識

Java NIO 異步非阻塞 Socket 通信實戰：Selector、Channel、Buffer 核心機制與服務端完整實現

編輯：Java 相關知識

JDK 1.4 引入 java.nio 包，徹底改寫 Java 網絡編程模型：從「每連接一線程」同步阻塞模型，轉向「單線程多路複用」異步非阻塞模型。本文深度解析 Selector、Channel、Buffer、SelectionKey 核心組件交互機制，並給出生產級非阻塞服務端完整代碼，經 Windows 2000 + J2SE 1.4 + Telnet 實測驗證。

同步阻塞 vs 異步非阻塞

維度	傳統 BIO	NIO
線程模型	每連接一線程	單/少線程處理成百上千連接
I/O 調用	阻塞直到完成	非阻塞立即返回，事件驅動
擴展性	線程開銷大，受限於 OS 線程數	高併發、低延遲、資源佔用少
編程複雜度	直觀簡單	狀態機/回調風格，較複雜

NIO 核心組件交互圖譜


┌─────────────┐     註冊感興趣事件      ┌─────────────┐
│  Channel    │ ─────────────────────▶ │  Selector   │
│ (ServerSocketChannel)                │ (多路複用器) │
└─────────────┘                        └──────┬──────┘
                                               │ 輪詢 ready keys
                                               ▼
┌─────────────┐     通知就緒事件       ┌─────────────┐
│  Channel    │ ◀───────────────────── │ SelectionKey │
│ (SocketChannel)                      │ (綁定 Channel+事件) │
└─────────────┘                        └─────────────┘
       │                                       │
       ▼                                       ▼
┌─────────────┐                        ┌─────────────┐
│   Buffer    │ ◀─── 讀/寫數據 ──────▶ │  應用邏輯    │
│ (ByteBuffer)│                        │  (解碼/業務)  │
└─────────────┘                        └─────────────┘

關鍵類職責

Channel：雙向通道，ServerSocketChannel/SocketChannel。configureBlocking(false) 切換非阻塞模式
Selector：多路複用器，select() 阻塞等待就緒事件，selectNow()/select(timeout) 非阻塞/定時輪詢。最多支持 63 通道同時註冊
SelectionKey：Channel 與 Selector 註冊關係的令牌。isAcceptable()/isReadable()/isWritable()/isConnectable() 判定就緒事件類型。attach(obj) 攜帶業務上下文
Buffer：數據容器，ByteBuffer 核心。flip() 切換讀模式，compact() 處理未讀完數據，clear() 重置寫模式
Charset/Decoder/Encoder：字節↔字符轉換，支持指定編碼

服務端核心邏輯流程

初始化：Selector.open()、ServerSocketChannel.open()、綁定端口、註冊 OP_ACCEPT
事件循環：while (selector.select() > 0) 輪詢就緒 Key 集合
分派處理：
- OP_ACCEPT：serverChannel.accept() 獲取 SocketChannel，設置非阻塞，註冊 OP_READ | OP_WRITE，attach(Socket) 攜帶 Socket 以便關閉，存入 Map
- OP_READ：從 Key 取 Channel/附件，讀入 ByteBuffer，解碼字符串，@exit 命令關閉 Channel，換行觸發業務處理（回調 ClientContext.execute()）
- OP_WRITE：向 Channel 寫入響應數據
資源釋放：finalize() 關閉 Channel/Selector

完整服務端代碼

import java.io.*; import java.net.*; import java.nio.*; import java.nio.channels.*; import java.nio.charset.*; import java.util.*;

public class NBlockingServer {
    int port = 8000, BUFFERSIZE = 1024;
    Selector selector;
    ServerSocketChannel serverChannel;
    Map clientMap = new HashMap<>();

    public NBlockingServer(int port) { this.port = port; }

    public void initialize() throws IOException {
        selector = Selector.open();
        serverChannel = ServerSocketChannel.open();
        serverChannel.configureBlocking(false);
        serverChannel.socket().bind(new InetSocketAddress(InetAddress.getLocalHost(), port));
    }

    public void finalize() throws IOException {
        serverChannel.close(); selector.close();
    }

    String decode(ByteBuffer bb) throws CharacterCodingException {
        CharsetDecoder dec = Charset.forName("ISO-8859-1").newDecoder();
        return dec.decode(bb).toString();
    }

    void writeToChannel(SocketChannel ch, String msg) throws IOException {
        ch.write(ByteBuffer.wrap(msg.getBytes()));
    }

    void readFromChannel(SocketChannel ch, ClientContext ctx) throws IOException, InterruptedException {
        ByteBuffer bb = ByteBuffer.allocate(BUFFERSIZE);
        int n = ch.read(bb);
        if (n == -1) { ch.close(); return; }
        bb.flip();
        String msg = decode(bb);
        if (msg.indexOf("@exit") >= 0) { ch.close(); return; }
        ctx.append(msg);
        if (msg.indexOf("\n") >= 0) {
            System.out.println("client input: " + msg);
            ctx.execute();
        }
    }

    public void portListening() throws IOException, InterruptedException {
        serverChannel.register(selector, SelectionKey.OP_ACCEPT);
        while (selector.select() > 0) {
            Set ready = selector.selectedKeys();
            for (Iterator it = ready.iterator(); it.hasNext(); ) {
                SelectionKey key = it.next(); it.remove();
                if (key.isAcceptable()) {
                    ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
                    SocketChannel sc = ssc.accept();
                    sc.configureBlocking(false);
                    SelectionKey rwKey = sc.register(selector, SelectionKey.OP_READ | SelectionKey.OP_WRITE);
                    rwKey.attach(sc.socket());
                    clientMap.put(sc.socket(), new ClientContext(sc));
                    System.out.println("New client connected");
                } else if (key.isReadable()) {
                    Socket socket = (Socket) key.attachment();
                    readFromChannel(socket.getChannel(), clientMap.get(socket));
                } else if (key.isWritable()) {
                    Socket socket = (Socket) key.attachment();
                    writeToChannel(socket.getChannel(), "Server echo: OK");
                }
            }
        }
    }

    // 客戶端上下文：緩衝未完整行、業務回調
    class ClientContext {
        SocketChannel channel; StringBuffer buffer = new StringBuffer();
        ClientContext(SocketChannel ch) { this.channel = ch; }
        void append(String s) { buffer.append(s); }
        void execute() throws IOException {
            writeToChannel(channel, "Response after read: " + buffer);
            buffer = new StringBuffer();
        }
    }

    public static void main(String[] args) {
        NBlockingServer srv = new NBlockingServer(8000);
        try { srv.initialize(); } catch (Exception e) { e.printStackTrace(); System.exit(-1); }
        try { srv.portListening(); } catch (Exception e) { e.printStackTrace(); }
    }
}