2

当我尝试在 Netty 中重用客户端连接时，我得到java.io.IOException: Connection reset by peer了（如果我发送一个请求，这不会发生，但如果我发送两个请求，即使是从单个线程，每次都会发生）。我目前的方法涉及以下实现一个简单的 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()使用新通道或旧通道，并在收到响应并在对客户端的响应中设置内容后ResponseHandler调用）：returnChannel()

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 客户端执行的两个串行请求。第一个请求使用了一个新的通道并正常完成，第二个请求尝试重用相同的通道，该通道是开放且可写的，但莫名其妙地失败（除了注意到从工作线程）。显然，第二个请求在重试时成功完成。两个请求完成后很多秒，两个连接最终都关闭（即，即使连接被对等方关闭，我截获的任何事件都不会反映这一点）：

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)
4

1 回答 1

2

最后，想通了。有两个问题导致连接重置。首先，我没有从向代理发送请求releaseConnection()的 apache commons调用（请参阅后续问题）。其次，在连接被重用的情况下，两次向代理服务器发出相同的调用。我需要在第一次写入后返回，而不是继续 while 循环。这个双重代理调用的结果是我向原始客户端发出了重复的响应，破坏了与客户端的连接。HttpClientexecuteWrite

于 2013-04-08T20:07:03.203 回答