java - Netty ServerBootStrap选项或childOption及其缓冲区大小对速度的影响

当缓冲区大小增加到默认值以上和/或在ServerBootStrap对象中选择了错误的替代位置(option或childOption)时，Netty似乎会降低上载/下载速度。当连接具有更大的延迟(〜300ms)时，这一点变得更加明显

设置:

Netty客户端在MacOS上，仅具有默认值。使用具有300ms DNS延迟的“网络链接调节器”。默认值:SendBuffer/ReceiveBuffer/LowWaterMark/HighWaterMark-128KB/128KB/32KB/64KB

Netty服务器在Windows 8.1上，其默认值为SendBuffer/ReceiveBuffer/LowWaterMark/HighWaterMark-64KB/64KB/32KB/64KB。

Netty版本4.1.6最终版。

使用具有以下设置的IO图，通过Wireshark测量速度:Y轴-> SUM(Y字段)，Y字段-> tcp.len

设备位于本地网络上。

结果(速度值):

服务器->客户端传输(将SO_SNDBUF和水印的低/高值设置为0.5 * SO_SNDBUF/SO_SNDBUF):

Setting location\SO_SNDBUF   | 64KB   | 128KB  | 1024KB
-----------------------------------------------------------------
option                       | 3.6MB/s| 3.6MB/s| 3.6MB/s
childOption                  | 0.2MB/s| 0.5MB/s| 3.6MB/s

客户端->服务器传输(设置SO_RCVBUF):

Setting location\SO_RCVBUF   | 64KB   | 128KB  | 1024KB
-----------------------------------------------------------------
option                       | 0.2MB/s| 0.5MB/s| 3.6MB/s
childOption                  | 3.6MB/s| 0.4MB/s| 3.6MB/s

服务器代码:

“参数”值:

“sendToClient”/“sendToServer”(隐式)用于传输方向。
“选项”/“childOption”表示替代类型。
“1”/“2”/“3”表示缓冲区值。

ObjectEchoServer.java

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslContextBuilder;
import io.netty.handler.ssl.util.SelfSignedCertificate;

public final class ObjectEchoServer {

    static final int PORT = 8007;

    public static void main(String[] args) throws Exception {
        int rcvBuf, sndBuf, lowWaterMark, highWaterMark;
        rcvBuf = sndBuf = lowWaterMark = highWaterMark = 0;
        switch (args[2]){
            case "1":
                rcvBuf = 64;
                sndBuf = 64;
                lowWaterMark = 32;
                highWaterMark = 64;
                break;
            case "2":
                rcvBuf = 128;
                sndBuf = 128;
                lowWaterMark = 64;
                highWaterMark = 128;
                break;
            case "3":
                rcvBuf = 1024;
                sndBuf = 1024;
                lowWaterMark = 512;
                highWaterMark = 1024;
                break;

        }

        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
             .channel(NioServerSocketChannel.class)
             .handler(new LoggingHandler(LogLevel.INFO))
             .childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                public void initChannel(SocketChannel ch) throws Exception {
                    ChannelPipeline p = ch.pipeline();
                    p.addLast(
                            new ObjectEncoder(),
                            new ObjectDecoder(ClassResolvers.cacheDisabled(null)),
                            new ObjectEchoServerHandler(args[0]));
                }
             });
            if(args[1].equalsIgnoreCase("childOption")) {
                b.childOption(ChannelOption.SO_RCVBUF, rcvBuf * 1024);
                b.childOption(ChannelOption.SO_SNDBUF, sndBuf * 1024);
                b.childOption(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark(lowWaterMark * 1024, highWaterMark * 1024));
            } else if(args[1].equalsIgnoreCase("option")){
                b.option(ChannelOption.SO_RCVBUF, rcvBuf * 1024);
                b.option(ChannelOption.SO_SNDBUF, sndBuf * 1024);
                b.option(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark(lowWaterMark * 1024, highWaterMark * 1024));
            }
            // Bind and start to accept incoming connections.
            b.bind(PORT).sync().channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

ObjectEchoServerHandler.java

import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

public class ObjectEchoServerHandler extends ChannelInboundHandlerAdapter {
    private Object msg;
    ChannelHandlerContext ctx;
    String sendToClient;

    public ObjectEchoServerHandler(String sendToClient){
        this.sendToClient = sendToClient;

    }

    @Override
    public void channelActive(ChannelHandlerContext ctx){
        this.ctx = ctx;
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        if(sendToClient.equalsIgnoreCase("sendToClient")) {//send a data stream to server
            this.msg = msg;
            ctx.writeAndFlush(msg).addListener(trafficGenerator);
        } //else receive a data stream from client
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) {
        ctx.flush();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }

    private final ChannelFutureListener trafficGenerator = new ChannelFutureListener() {
        @Override
        public void operationComplete(ChannelFuture future) {
            if (future.isSuccess()) {
                ctx.writeAndFlush(msg).addListener(trafficGenerator);
            } else {
                future.cause().printStackTrace();
                future.channel().close();
            }
        }
    };
}

客户代码:

“参数”值:

“sendToClient”/“sendToServer”(隐式)用于传输方向。

ObjectEchoClient.java

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;

public final class ObjectEchoClient {

    static final String HOST = System.getProperty("host", "127.0.0.1");
    static final int PORT = 8007;
    static final int SIZE = 256;

    public static void main(String[] args) throws Exception {

        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(group)
             .channel(NioSocketChannel.class)
             .handler(new ChannelInitializer<SocketChannel>() {
                @Override
                public void initChannel(SocketChannel ch) throws Exception {
                    ChannelPipeline p = ch.pipeline();
                    p.addLast(
                            new ObjectEncoder(),
                            new ObjectDecoder(ClassResolvers.cacheDisabled(null)),
                            new ObjectEchoClientHandler(args[0]));
                    System.out.println("senbuf:"+ ch.config().getSendBufferSize());
                    System.out.println("waterhigh:"+ ch.config().getWriteBufferWaterMark().high());
                    System.out.println("waterlow:"+ ch.config().getWriteBufferWaterMark().low());
                    System.out.println("recbuf:"+ ch.config().getReceiveBufferSize());

                }
             });
            // Start the connection attempt.
            b.connect(HOST, PORT).sync().channel().closeFuture().sync();
        } finally {
            group.shutdownGracefully();
        }
    }
}

ObjectEchoClientHandler.java

import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

import java.util.ArrayList;
import java.util.List;

public class ObjectEchoClientHandler extends ChannelInboundHandlerAdapter {

    private final List<String> firstMessage;
    private ChannelHandlerContext ctx;
    private String sendToClient;

    ObjectEchoClientHandler(String sendToClient) {
        this.sendToClient = sendToClient;

        firstMessage = new ArrayList<>(ObjectEchoClient.SIZE);
        for (int i = 0; i < ObjectEchoClient.SIZE; i++) {
            firstMessage.add(Integer.toString(i));
        }
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        this.ctx = ctx;
      if(sendToClient.equalsIgnoreCase("sendToClient")) {//get a data stream from server
          this.ctx.writeAndFlush(firstMessage);
      } else {//send a stream of to server
          this.ctx.writeAndFlush(firstMessage).addListener(trafficGenerator);
      }
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        ctx.flush();
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) {
        ctx.flush();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }

    private final ChannelFutureListener trafficGenerator = new ChannelFutureListener() {
        @Override
        public void operationComplete(ChannelFuture future) {
            if (future.isSuccess()) {
                ctx.writeAndFlush(firstMessage).addListener(trafficGenerator);
            } else {
                future.cause().printStackTrace();
                future.channel().close();
            }
        }
    };
}

大问题:

设置缓冲区的正确方法/位置是什么？我发现的信息(主要是代码示例)到处都是。

一个线程(What is the difference between ServerBootstrap.option() and ServerBootstrap.childOption() in netty 4.x)表示应将ServerBootStrap.childOption用于每个客户端信息，因此我认为两个缓冲区都应位于ServerBootStrap.childOption中。

最佳答案

我认为这是对的b.option(ChannelOption.SO_RCVBUF, rcvBuf * 1024); b.option(ChannelOption.SO_SNDBUF, sndBuf * 1024);