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

Java NIO TCP 編程實戰：Selector 模式實現高性能 Echo 服務端與客戶端

編輯：Java 相關知識

NIO 核心組件回顧

Channel：雙向通道（SocketChannel、ServerSocketChannel），可非阻塞
Buffer：數據容器（ByteBuffer），支援讀寫模式切換
Selector：多路復用器，單線程管理成百上千 Channel
SelectionKey：Channel 與 Selector 註冊關係，包含就緒操作位掩碼

協議接口設計：關注點分離

public interface EchoProtocol {
    void handleAccept(SelectionKey key) throws IOException;
    void handleRead(SelectionKey key) throws IOException;
    void handleWrite(SelectionKey key) throws IOException;
}

TCP 協議實現：狀態機驅動

public class TCPEchoSelectorProtocol implements EchoProtocol {
    private final int bufSize;
    public TCPEchoSelectorProtocol(int bufSize) { this.bufSize = bufSize; }

    @Override
    public void handleAccept(SelectionKey key) throws IOException {
        ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
        SocketChannel client = ssc.accept();
        client.configureBlocking(false); // 必須非阻塞
        // 註冊讀事件，附帶讀緩衝區
        client.register(key.selector(), SelectionKey.OP_READ, ByteBuffer.allocate(bufSize));
    }

    @Override
    public void handleRead(SelectionKey key) throws IOException {
        SocketChannel client = (SocketChannel) key.channel();
        ByteBuffer buf = (ByteBuffer) key.attachment();
        long read = client.read(buf);
        if (read == -1) { client.close(); return; } // 對端關閉
        if (read > 0) {
            // 數據就緒，同時關注寫事件
            key.interestOps(SelectionKey.OP_READ | SelectionKey.OP_WRITE);
        }
    }

    @Override
    public void handleWrite(SelectionKey key) throws IOException {
        ByteBuffer buf = (ByteBuffer) key.attachment();
        buf.flip(); // 切換讀模式
        SocketChannel client = (SocketChannel) key.channel();
        client.write(buf);
        if (!buf.hasRemaining()) { // 寫完畢
            key.interestOps(SelectionKey.OP_READ); // 只關注讀
        }
        buf.compact(); // 整理緩衝區，為下次讀取騰出空間
    }
}

非阻塞客戶端：主動發起連接與輪詢

public class TCPEchoClientNonblocking {
    public static void main(String[] args) throws Exception {
        SocketChannel client = SocketChannel.open();
        client.configureBlocking(false);
        // 非阻塞連接：發起後輪詢 finishConnect
        if (!client.connect(new InetSocketAddress("127.0.0.1", 5500))) {
            while (!client.finishConnect()) { System.out.print("."); }
        }
        ByteBuffer writeBuf = ByteBuffer.wrap("0123456789abcdefghijklmnopqrstuvwxyz".getBytes());
        ByteBuffer readBuf = ByteBuffer.allocate(writeBuf.capacity());
        int total = 0;
        while (total < writeBuf.capacity()) {
            if (writeBuf.hasRemaining()) client.write(writeBuf);
            int r = client.read(readBuf);
            if (r == -1) throw new SocketException("連接提前關閉");
            total += r;
            System.out.print(".");
        }
        System.out.println("收到: " + new String(readBuf.array(), 0, total).length());
        client.close();
    }
}

Selector 服務端主循環

public class TCPServerSelector {
    private static final int BUFSIZE = 256, TIMEOUT = 3000;
    public static void main(String[] args) throws IOException {
        Selector selector = Selector.open();
        // 註冊監聽通道
        ServerSocketChannel listen = ServerSocketChannel.open();
        listen.socket().bind(new InetSocketAddress(5500));
        listen.configureBlocking(false);
        listen.register(selector, SelectionKey.OP_ACCEPT);

        EchoProtocol protocol = new TCPEchoSelectorProtocol(BUFSIZE);
        while (true) {
            if (selector.select(TIMEOUT) == 0) { System.out.print("."); continue; }
            Iterator it = selector.selectedKeys().iterator();
            while (it.hasNext()) {
                SelectionKey key = it.next(); it.remove();
                if (key.isAcceptable()) { protocol.handleAccept(key); }
                else if (key.isReadable()) { protocol.handleRead(key); }
                else if (key.isValid() && key.isWritable()) { protocol.handleWrite(key); }
            }
        }
    }
}