Netty 代理延迟响应

Question

我正在使用 Netty 框架创建代理，但我注意到收到的最后一条消息在传递到下一个节点之前往往会延迟。

设计：

客户||代理 ||服务器

基本上，当服务器启动一条消息时，问题就出现了，它们是在将消息传递给客户端之前的延迟，或者如果服务器在第一条消息之后立即发送后续消息，则第一条消息通过，第二条消息延迟几秒钟.为什么会这样？是否有一些我缺少的配置参数？

Startup.java

        Executor executor = Executors.newCachedThreadPool();
        ServerBootstrap sb = new ServerBootstrap(
                new NioServerSocketChannelFactory(executor, executor));

        // Set up the event pipeline factory.
        ClientSocketChannelFactory cf =
                new NioClientSocketChannelFactory(executor, executor);

        sb.setPipelineFactory(
                new ProxyPipelineFactory(cf, remoteHost, remotePort));

        sb.setOption("child.tcpNoDelay", true);
        sb.setOption("child.keepAlive", true);

        // Start up the server.
        sb.bind(new InetSocketAddress(localPort));

ProxyPipelineFactory.java

@Override
public ChannelPipeline getPipeline() throws Exception {
    ChannelPipeline p = pipeline(); 
    p.addLast("handler", new ClientHandler(cf, remoteHost, remotePort));
    return p;
}

ClientHandler.java

@Override
public void channelOpen(ChannelHandlerContext ctx, ChannelStateEvent e)
        throws Exception {
    // Suspend incoming traffic until connected to the remote host.
    final Channel inboundChannel = e.getChannel();
    inboundChannel.setReadable(false);

    // Start the connection attempt.
    ClientBootstrap cb = new ClientBootstrap(cf);

        cb.setOption("child.tcpNoDelay", true);
        cb.setOption("child.keepAlive", true);
    ChannelPipeline p = cb.getPipeline();
    p.addLast("famer", new DelimiterBasedFrameDecoder(8192, false, new ChannelBuffer[]{ChannelBuffers.wrappedBuffer("</cmd>".getBytes())}));
p.addLast("handler", new ServerHandler(e.getChannel(), trafficLock));
    ChannelFuture f = cb.connect(new InetSocketAddress(remoteHost, remotePort));

    outboundChannel = f.getChannel();
    f.addListener(new ChannelFutureListener() {

        @Override
        public void operationComplete(ChannelFuture future) throws Exception {
            if (future.isSuccess()) {
                // Connection attempt succeeded:
                // Begin to accept incoming traffic.
                inboundChannel.setReadable(true);
            } else {
                // Close the connection if the connection attempt has failed.
                inboundChannel.close();
            }
        }
    });
}

@Override
public void messageReceived(ChannelHandlerContext ctx, final MessageEvent e)
        throws Exception {

    BigEndianHeapChannelBuffer msg = (BigEndianHeapChannelBuffer) e.getMessage();

    if (log.isDebugEnabled()) {
        byte[] bytes = new byte[msg.capacity()];
        msg.readBytes(bytes);
        msg.setIndex(0, bytes.length);
        StringBuilder out = new StringBuilder("\nPROXY[ ").append(e.getChannel().getRemoteAddress()).append(" ---> Server ]");
        out.append("\nMESSAGE length=").append(bytes.length).append("\n").append(new String(bytes));
        log.debug(out.toString());
    }

    synchronized (trafficLock) {
        outboundChannel.write(msg);
        // If outboundChannel is saturated, do not read until notified in
        // OutboundHandler.channelInterestChanged().
        if (!outboundChannel.isWritable()) {
            e.getChannel().setReadable(false);
        }
    }
}

ServerHandler.java

@Override
public void messageReceived(ChannelHandlerContext ctx, final MessageEvent e)
        throws Exception {
    BigEndianHeapChannelBuffer msg = (BigEndianHeapChannelBuffer) e.getMessage();
    proxy(e.getChannel(), msg);
}
private void proxy(Channel connection, ChannelBuffer raw) {
    synchronized (trafficLock) {
        inboundChannel.write(raw);
        // If inboundChannel is saturated, do not read until notified in
        // ClientHandler.channelInterestChanged().
        if (!inboundChannel.isWritable()) {
            connection.setReadable(false);
        }
    }
}

Answer 1

您禁用了连接的可读性，您在哪里再次打开它以读取更多字节？您可以阅读 netty.io 上的 Prxoy Server 示例，它的作用完全相同。

private void proxy(Channel connection, ChannelBuffer raw) {
    synchronized (trafficLock) {
        inboundChannel.write(raw);
        // If inboundChannel is saturated, do not read until notified in
        // ClientHandler.channelInterestChanged().
        if (!inboundChannel.isWritable()) {
            connection.setReadable(false);
        }
    }
}