Я пытаюсь создать простой Echo Service с использованием Java NIO 2 (который основан на шаблоне Proactor).
В простейшей реализации у нас есть 4 основных компонента; ProactorInitiator, AcceptConnectionHandler, ReadConnectionHandler и WriteConnectionHandler.
Ниже приведен пример кода.
ProactorInitiator.java
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.AsynchronousServerSocketChannel;
public class ProactorInitiator {
static int ASYNC_SERVER_PORT = 4333;
public void initiateProactiveServer(int port)
throws IOException {
final AsynchronousServerSocketChannel listener =
AsynchronousServerSocketChannel.open().bind(
new InetSocketAddress(port));
AcceptCompletionHandler acceptCompletionHandler =
new AcceptCompletionHandler(listener);
SessionState state = new SessionState();
listener.accept(state, acceptCompletionHandler);
System.out.println("Proactor Initiator Running on "+Thread.currentThread().getName());
}
public static void main(String[] args) {
try {
System.out.println("Async server listening on port : " +
ASYNC_SERVER_PORT);
new ProactorInitiator().initiateProactiveServer(
ASYNC_SERVER_PORT);
} catch (IOException e) {
e.printStackTrace();
}
// Sleep indefinitely since otherwise the JVM would terminate
while (true) {
try {
Thread.sleep(Long.MAX_VALUE);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
AcceptCompletionHandler.java
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousServerSocketChannel;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
public class AcceptCompletionHandler
implements
CompletionHandler<AsynchronousSocketChannel, SessionState> {
private AsynchronousServerSocketChannel listener;
public AcceptCompletionHandler(
AsynchronousServerSocketChannel listener) {
this.listener = listener;
}
@Override
public void completed(AsynchronousSocketChannel socketChannel,
SessionState sessionState) {
System.out.println("Accept Handler running on "+Thread.currentThread().getName());
// accept the next connection
SessionState newSessionState = new SessionState();
listener.accept(newSessionState, this);
// handle this connection
ByteBuffer inputBuffer = ByteBuffer.allocate(2048);
ReadCompletionHandler readCompletionHandler =
new ReadCompletionHandler(socketChannel, inputBuffer);
socketChannel.read(
inputBuffer, sessionState, readCompletionHandler);
}
@Override
public void failed(Throwable exc, SessionState sessionState) {
// Handle connection failure...
}
}
ReadCompletionHandler.java
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
public class ReadCompletionHandler implements
CompletionHandler<Integer, SessionState> {
private AsynchronousSocketChannel socketChannel;
private ByteBuffer inputBuffer;
public ReadCompletionHandler(
AsynchronousSocketChannel socketChannel,
ByteBuffer inputBuffer) {
this.socketChannel = socketChannel;
this.inputBuffer = inputBuffer;
}
@Override
public void completed(
Integer bytesRead, SessionState sessionState) {
System.out.println("Read Handler running on "+Thread.currentThread().getName());
byte[] buffer = new byte[bytesRead];
inputBuffer.rewind();
// Rewind the input buffer to read from the beginning
inputBuffer.get(buffer);
String message = new String(buffer);
// System.out.println("Received message from client : " + message);
// message = GetRequestParser.getHTTPRequest(message, "200 OK");
// Echo the message back to client
WriteCompletionHandler writeCompletionHandler =
new WriteCompletionHandler(socketChannel);
ByteBuffer outputBuffer = ByteBuffer.wrap(message.getBytes());
socketChannel.write(
outputBuffer, sessionState, writeCompletionHandler);
}
@Override
public void failed(Throwable exc, SessionState attachment) {
//Handle read failure.....
}
}
WriteCompletionHandler.java
import java.io.IOException;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
public class WriteCompletionHandler implements
CompletionHandler<Integer, SessionState> {
private AsynchronousSocketChannel socketChannel;
public WriteCompletionHandler(
AsynchronousSocketChannel socketChannel) {
this.socketChannel = socketChannel;
}
@Override
public void completed(
Integer bytesWritten, SessionState attachment) {
try {
System.out.println("Write Handler running on "+Thread.currentThread().getName());
System.out.println("\n");
socketChannel.close();
} catch (IOException e) {
e.printStackTrace();
}
}
@Override
public void failed(Throwable exc, SessionState attachment) {
// Handle write failure.....
}
}
SessionState.java
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
public class SessionState {
private Map<String, String> sessionProps =
new ConcurrentHashMap<>();
public String getProperty(String key) {
return sessionProps.get(key);
}
public void setProperty(String key, String value) {
sessionProps.put(key, value);
}
}
Чтобы проверить поведение потоков, я печатаю поток, в котором работает каждый обработчик, до 'sys.out'.
Ниже приведены различные результаты, которые я получил, для многих запросов, которые отправляются на сервер один за другим.
Запрос 1
Accept Handler running on Thread-4
Read Handler running on Thread-4
Write Handler running on Thread-4
Запрос 2
Accept Handler running on Thread-4
Read Handler running on Thread-2
Write Handler running on Thread-2
Запрос 3
Accept Handler running on Thread-5
Read Handler running on Thread-3
Write Handler running on Thread-3
Согласно приведенным выше результатам, кажется, что для разных запросов сервер использует разные потоки. Кроме того, оба обработчика чтения и записи выполняются в одном потоке для данного запроса.
Может кто-нибудь объяснить этот результат? Как планируются обработчики в разных потоках?