Netty でクライアント接続を再利用しようとすると、エラーがjava.io.IOException: Connection reset by peer発生します (これは、1 つのリクエストを送信した場合には発生しませんが、1 つのスレッドからでも 2 つのリクエストを送信すると毎回発生します)。私の現在のアプローチには、コードが以下の単純な ChannelPool の実装が含まれます。key メソッドは、メンバーから空きチャネルを取得するfreeChannelsか、使用可能なチャネルがない場合は新しいチャネルを作成することに注意してください。メソッドreturnChannel()は、リクエストの処理が完了したときにチャネルを解放するメソッドです。レスポンスを処理した後、パイプライン内で呼び出されます (以下のコードのmessageReceived()メソッドを参照)。ResponseHandler私が間違っていること、なぜ例外が発生するのか、誰かがわかりますか?
チャネル プール コード (freeChannels.pollFirst()への呼び出しによって返された空きチャネルを取得するために を使用することに注意してくださいreturnChannel()):
public class ChannelPool {
private final ClientBootstrap cb;
private Deque<Channel> freeChannels = new ArrayDeque<Channel>();
private static Map<Channel, Channel> proxyToClient = new ConcurrentHashMap<Channel, Channel>();
public ChannelPool(InetSocketAddress address, ChannelPipelineFactory pipelineFactory) {
ChannelFactory clientFactory =
new NioClientSocketChannelFactory(
Executors.newCachedThreadPool(),
Executors.newCachedThreadPool());
cb = new ClientBootstrap(clientFactory);
cb.setPipelineFactory(pipelineFactory);
}
private void writeToNewChannel(final Object writable, Channel clientChannel) {
ChannelFuture cf;
synchronized (cb) {
cf = cb.connect(new InetSocketAddress("localhost", 18080));
}
final Channel ch = cf.getChannel();
proxyToClient.put(ch, clientChannel);
cf.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture arg0) throws Exception {
System.out.println("channel open, writing: " + ch);
ch.write(writable);
}
});
}
public void executeWrite(Object writable, Channel clientChannel) {
synchronized (freeChannels) {
while (!freeChannels.isEmpty()) {
Channel ch = freeChannels.pollFirst();
System.out.println("trying to reuse channel: " + ch + " " + ch.isOpen());
if (ch.isOpen()) {
proxyToClient.put(ch, clientChannel);
ch.write(writable).addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture cf) throws Exception {
System.out.println("write from reused channel complete, success? " + cf.isSuccess());
}
});
// EDIT: I needed a return here
}
}
}
writeToNewChannel(writable, clientChannel);
}
public void returnChannel(Channel ch) {
synchronized (freeChannels) {
freeChannels.addLast(ch);
}
}
public Channel getClientChannel(Channel proxyChannel) {
return proxyToClient.get(proxyChannel);
}
}
Netty パイプライン コード (新しいチャネルまたは古いチャネルのいずれかを使用する呼び出し、およびRequestHandler応答が受信され、コンテンツがクライアントへの応答に設定された後の呼び出し) :executeWrite()ResponseHandlerreturnChannel()
public class NettyExample {
private static ChannelPool pool;
public static void main(String[] args) throws Exception {
pool = new ChannelPool(
new InetSocketAddress("localhost", 18080),
new ChannelPipelineFactory() {
public ChannelPipeline getPipeline() {
return Channels.pipeline(
new HttpRequestEncoder(),
new HttpResponseDecoder(),
new ResponseHandler());
}
});
ChannelFactory factory =
new NioServerSocketChannelFactory(
Executors.newCachedThreadPool(),
Executors.newCachedThreadPool());
ServerBootstrap sb = new ServerBootstrap(factory);
sb.setPipelineFactory(new ChannelPipelineFactory() {
public ChannelPipeline getPipeline() {
return Channels.pipeline(
new HttpRequestDecoder(),
new HttpResponseEncoder(),
new RequestHandler());
}
});
sb.setOption("child.tcpNoDelay", true);
sb.setOption("child.keepAlive", true);
sb.bind(new InetSocketAddress(2080));
}
private static class ResponseHandler extends SimpleChannelHandler {
@Override
public void messageReceived(ChannelHandlerContext ctx, final MessageEvent e) {
final HttpResponse proxyResponse = (HttpResponse) e.getMessage();
final Channel proxyChannel = e.getChannel();
Channel clientChannel = pool.getClientChannel(proxyChannel);
HttpResponse clientResponse = new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK);
clientResponse.setHeader(HttpHeaders.Names.CONTENT_TYPE, "text/html; charset=UTF-8");
HttpHeaders.setContentLength(clientResponse, proxyResponse.getContent().readableBytes());
clientResponse.setContent(proxyResponse.getContent());
pool.returnChannel(proxyChannel);
clientChannel.write(clientResponse);
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) {
e.getCause().printStackTrace();
Channel ch = e.getChannel();
ch.close();
}
}
private static class RequestHandler extends SimpleChannelHandler {
@Override
public void messageReceived(ChannelHandlerContext ctx, final MessageEvent e) {
final HttpRequest request = (HttpRequest) e.getMessage();
pool.executeWrite(request, e.getChannel());
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) {
e.getCause().printStackTrace();
Channel ch = e.getChannel();
ch.close();
}
}
}
編集: 詳細を説明するために、プロキシ接続で何が起こっているかのトレースを書きました。以下には、同期 apache commons クライアントによって実行される 2 つのシリアル リクエストが含まれることに注意してください。最初のリクエストは新しいチャネルを使用して正常に完了し、2 番目のリクエストは開いていて書き込み可能な同じチャネルを再利用しようとしますが、不可解に失敗します (私は、ワーカー スレッド)。明らかに、再試行が行われると、2 番目の要求は正常に完了します。両方の要求が完了してから数秒後、両方の接続が最終的に閉じられます (つまり、接続がピアによって閉じられたとしても、これは私が傍受したイベントには反映されません)。
channel open: [id: 0x6e6fbedf]
channel connect requested: [id: 0x6e6fbedf]
channel open, writing: [id: 0x6e6fbedf, /127.0.0.1:47031 => localhost/127.0.0.1:18080]
channel connected: [id: 0x6e6fbedf, /127.0.0.1:47031 => localhost/127.0.0.1:18080]
trying to reuse channel: [id: 0x6e6fbedf, /127.0.0.1:47031 => localhost/127.0.0.1:18080] true
channel open: [id: 0x3999abd1]
channel connect requested: [id: 0x3999abd1]
channel open, writing: [id: 0x3999abd1, /127.0.0.1:47032 => localhost/127.0.0.1:18080]
channel connected: [id: 0x3999abd1, /127.0.0.1:47032 => localhost/127.0.0.1:18080]
java.io.IOException: Connection reset by peer
at sun.nio.ch.FileDispatcherImpl.read0(Native Method)
at sun.nio.ch.SocketDispatcher.read(SocketDispatcher.java:39)
at sun.nio.ch.IOUtil.readIntoNativeBuffer(IOUtil.java:218)
at sun.nio.ch.IOUtil.read(IOUtil.java:186)
at sun.nio.ch.SocketChannelImpl.read(SocketChannelImpl.java:359)
at org.jboss.netty.channel.socket.nio.NioWorker.read(NioWorker.java:63)
at org.jboss.netty.channel.socket.nio.AbstractNioWorker.processSelectedKeys(AbstractNioWorker.java:373)
at org.jboss.netty.channel.socket.nio.AbstractNioWorker.run(AbstractNioWorker.java:247)
at org.jboss.netty.channel.socket.nio.NioWorker.run(NioWorker.java:35)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1110)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:603)
at java.lang.Thread.run(Thread.java:722)