Deep-dive: Non-Blocking Server trong Java (NIO, Selector, Buffer, Channel)
Cách một thread phục vụ nhiều kết nối cùng lúc bằng Java NIO — Selector, Buffer, Channel, và 8 bước dựng một non-blocking server từ đầu.
Tổng quan
Non-blocking server (server không chặn) là server có thể xử lý nhiều request đồng thời trong cùng một process hoặc thậm chí cùng một thread, nhờ dùng non-blocking I/O (vào/ra không chặn). Theo cách tiếp cận này, một thread có thể phục vụ nhiều truy vấn cùng lúc.
Server dùng non-blocking socket hoạt động theo kiểu asynchronous (bất đồng bộ): khi nhận được request, nó giao việc cho hệ thống xử lý rồi tiếp tục nhận các request khác; khi nào việc xong thì mới phản hồi cho từng request tương ứng. Nó không đứng chờ.
Bài này đi qua toàn bộ cách tự dựng một non-blocking server và client bằng Java NIO.
Selector — bộ điều phối kênh
Selector là một trong các class cốt lõi của Java NIO. Mọi channel (kênh) mà ta muốn lắng nghe đều phải được register (đăng ký) vào Selector.
Mỗi channel khi đăng ký sẽ được gán một SelectionKey — đối tượng định danh channel đó, đồng thời mang thông tin về trạng thái của nó, ví dụ channel đã sẵn sàng để nhận kết nối hay chưa. Nó cho biết loại request là gì và ai đang gửi request.
Điểm quan trọng: một instance Selector duy nhất có thể theo dõi rất nhiều socket channel, và tự thông báo cho ứng dụng biết channel nào đã sẵn sàng để xử lý. Đây chính là thứ cho phép một thread phục vụ nhiều kết nối.
Selector selector = Selector.open();Buffer — vùng đệm dữ liệu
Buffer (vùng đệm) định nghĩa các thuộc tính chung cho mọi loại đệm: limit (giới hạn), capacity (dung lượng) và position (vị trí hiện tại). Buffer trong Java NIO là thứ dùng để tương tác với channel.
Bản chất buffer là một khối bộ nhớ (block of memory) mà ta ghi dữ liệu vào, rồi sau đó đọc lại. Khối bộ nhớ đó được bọc trong một đối tượng buffer của NIO, giúp làm việc với nó dễ hơn là thao tác bộ nhớ thô.
Mỗi kiểu dữ liệu nguyên thuỷ có một loại buffer riêng:
ByteBufferCharBufferShortBufferIntBufferLongBufferFloatBufferDoubleBuffer
Cấp phát một ByteBuffer dung lượng 1024 byte:
ByteBuffer buffer = ByteBuffer.allocate(1024);Các class implement Java NIO channel
SocketChannel
SocketChannel dùng để nối một channel với một TCP network socket. Có hai tình huống tạo ra SocketChannel:
- Khi có kết nối đến (incoming connection) tới
ServerSocketChannel. - Khi ta chủ động kết nối tới một server nào đó trên internet.
Mở bằng SocketChannel.open():
SocketChannel socketChannel =
SocketChannel.open(new InetSocketAddress("localhost", 8089));ServerSocketChannel
ServerSocketChannel cũng dùng để nối channel với TCP network socket, nhưng ở phía server. Tạo bằng open().
Lưu ý một cái bẫy: channel vừa tạo ra thì đã mở nhưng chưa bind (chưa gắn vào cổng nào). Gọi accept() trên một channel chưa bind sẽ ném NotYetBoundException. Phải bind trước:
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
serverSocketChannel.socket().bind(new InetSocketAddress("localhost", 8089));
while (true) {
SocketChannel socketChannel = serverSocketChannel.accept();
}8 bước dựng một non-blocking server
Bước 1 — Mở Selector để quản lý channel
Selector giữ tham chiếu tới một tập channel, và khi được hỏi thì trả về những channel đang sẵn sàng.
selector = Selector.open();Bước 2 — Bind cổng của server
ServerSocketChannel socket = ServerSocketChannel.open();
socket.configureBlocking(false);
ServerSocket serverSocket = socket.socket();
serverSocket.bind(new InetSocketAddress("localhost", 8089));Chú ý configureBlocking(false) — đây là dòng biến channel thành non-blocking. Thiếu nó thì mọi thứ còn lại vô nghĩa.
Bước 3 — Đăng ký ServerSocket vào Selector
socket.register(selector, ops, null);Bước 4 — Chờ sự kiện
Vào vòng lặp vô hạn và gọi select(). Hàm này chọn ra tập các key mà channel tương ứng đã sẵn sàng cho thao tác I/O.
Điều nghe hơi trái ngược: select() là một thao tác blocking (chặn) — nó chỉ trả về khi có ít nhất một channel được chọn. Nhưng đây là chỗ duy nhất chương trình chặn, và nó chặn để chờ bất kỳ channel nào trong hàng nghìn channel. Đó chính là điều làm nên hiệu quả: một thread chặn một lần, thay vì nghìn thread mỗi thread chặn một chỗ.
selector.select();Bước 5 — Lấy các selection key
Set<SelectionKey> selectedKeys = selector.selectedKeys();
Iterator<SelectionKey> i = selectedKeys.iterator();Bước 6 — Kiểm tra client đã sẵn sàng để accept chưa
isAcceptable() kiểm tra xem channel của key đó có sẵn sàng nhận một kết nối socket mới hay không. Nó tương đương hoàn toàn với biểu thức key.readyOps() & OP_ACCEPT != 0.
if (key.isAcceptable()) {
// Đã có client mới được accept
}Bước 7 — Đọc message của client
Sau khi kết nối được accept, ta đăng ký channel cho thao tác read, vì client sẽ gửi message tới server.
client.register(selector, SelectionKey.OP_READ);Dữ liệu sẽ được đọc từ channel vào buffer, rồi từ buffer ta lấy ra để in.
Bước 8 — Đóng kết nối
Đóng kết nối tới client bằng SocketChannel.close().
DEMO
File: Server.java
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
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;
public class Server {
private static Selector selector = null;
public static void main(String[] args) {
try {
selector = Selector.open();
// Thiết lập host, port và chuyển sang chế độ non-blocking
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
ServerSocket serverSocket = serverSocketChannel.socket();
serverSocket.bind(new InetSocketAddress("localhost", 8089));
serverSocketChannel.configureBlocking(false);
int ops = serverSocketChannel.validOps();
serverSocketChannel.register(selector, ops, null);
while (true) {
selector.select();
Set<SelectionKey> selectedKeys = selector.selectedKeys();
Iterator<SelectionKey> i = selectedKeys.iterator();
while (i.hasNext()) {
SelectionKey key = i.next();
if (key.isAcceptable()) {
// Có client mới
handleAccept(serverSocketChannel, key);
} else if (key.isReadable()) {
// Đọc dữ liệu từ client, không chặn
handleRead(key);
}
i.remove();
}
}
} catch (IOException e) {
e.printStackTrace();
}
}
private static void handleAccept(ServerSocketChannel mySocket, SelectionKey key)
throws IOException {
System.out.println("Connection Accepted..");
// Accept kết nối và chuyển sang non-blocking
SocketChannel client = mySocket.accept();
client.configureBlocking(false);
// Đăng ký để theo dõi sự kiện đọc trên channel này
client.register(selector, SelectionKey.OP_READ);
}
private static void handleRead(SelectionKey key) throws IOException {
System.out.println("Reading client's message.");
// Lấy lại channel từ key
SocketChannel client = (SocketChannel) key.channel();
// Tạo buffer để đọc dữ liệu
ByteBuffer buffer = ByteBuffer.allocate(1024);
int bytesRead = client.read(buffer); // Lấy giá trị trả về của read()
if (bytesRead == -1) {
// Client đã đóng kết nối
System.out.println("Client disconnected.");
client.close(); // Đóng SocketChannel
key.cancel(); // Hủy SelectionKey
System.out.println("Connection closed...");
} else if (bytesRead > 0) {
// Có dữ liệu để đọc
buffer.flip(); // Chuẩn bị buffer để đọc từ nó
// Chuyển dữ liệu từ buffer thành String
String data = new String(buffer.array()).trim();
if (!data.isEmpty()) {
System.out.println("Received message: " + data);
}
}
}
}File: Client.java
Phía client dùng virtual thread (luồng ảo, có từ Java 21) để bắn 10 request song song mà không cần thread pool cố định:
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class Client {
public static void main(String[] args) {
int numberOfTasks = 10;
// Tạo một ExecutorService sẽ tạo một virtual thread mới cho mỗi tác vụ.
// Đây là cách khuyến nghị để sử dụng virtual threads cho các tác vụ ngắn hạn.
try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
System.out.println("Submitting " + numberOfTasks
+ " tasks to the virtual thread executor...");
for (int i = 0; i < numberOfTasks; i++) {
final int taskNumber = i + 1; // Để dễ theo dõi trong log
executor.submit(() -> {
System.out.println("Starting task " + taskNumber);
sendRequest();
System.out.println("Finished task " + taskNumber);
});
}
System.out.println("All tasks submitted. Waiting for completion...");
}
// Khối try-with-resources sẽ tự động gọi executor.close(), tương đương
// shutdown() + awaitTermination(...) cho newVirtualThreadPerTaskExecutor.
// Nó sẽ chờ tất cả các virtual thread đã được submit hoàn thành.
System.out.println("All virtual threads have completed their tasks.");
}
private static void sendRequest() {
try {
String[] messages = {
"Non-Blocking servers are the best.",
"I Love LinhCN",
"I Love CNLinh"
};
System.out.println("Connection accepted by the Server..");
SocketChannel client = SocketChannel.open(
new InetSocketAddress("localhost", 8089));
for (String msg : messages) {
ByteBuffer buffer = ByteBuffer.allocate(1024);
buffer.put(msg.getBytes());
buffer.flip();
int bytesWritten = client.write(buffer);
System.out.println(String.format(
"Sending Message: %s\nbufferBytes: %d", msg, bytesWritten));
}
client.close();
System.out.println("Client connection closed");
} catch (IOException e) {
e.printStackTrace();
}
}
}Vì sao không thể tin vào ranh giới của message
Đây là phần dễ sai nhất khi làm việc với TCP và non-blocking I/O.
TCP là giao thức truyền theo luồng byte (stream-based), nó không giữ lại ranh giới của các “message” ban đầu khi gửi đi. Khi client gọi client.write(buffer) nhiều lần liên tiếp, hệ điều hành không đảm bảo rằng mỗi lần gọi write sẽ tương ứng với một lần read duy nhất ở phía server. Các byte từ những lần write khác nhau có thể:
- được gộp lại với nhau trong buffer của kernel TCP trước khi được gửi qua mạng, hoặc
- cùng đến buffer nhận của server và sẵn sàng để đọc trong cùng một lúc.
Về read() ở chế độ non-blocking: khi server gọi client.read(buffer) trên một channel non-blocking, lệnh này đọc tất cả các byte hiện đang có sẵn trong bộ đệm nhận của kernel cho kết nối đó, tối đa bằng kích thước buffer được cấp. Nó trả về ngay lập tức, kèm số byte đã đọc — có thể là 0, -1, hoặc một số dương.
Hệ quả thực tế: chạy demo ở trên, ba message "Non-Blocking servers are the best.", "I Love LinhCN", "I Love CNLinh" hoàn toàn có thể đến server dính liền thành một chuỗi trong cùng một lần read(). Muốn tách đúng từng message, ta phải tự định nghĩa framing (đóng khung dữ liệu) ở tầng application — ví dụ mỗi message có 4 byte độ dài ở đầu, hoặc kết thúc bằng một ký tự phân cách (delimiter).
Đây chính là lý do các framework như Netty tồn tại: chúng cung cấp sẵn LengthFieldBasedFrameDecoder, DelimiterBasedFrameDecoder… để ta không phải viết lại phần này mỗi lần.