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

Question

当缓冲区大小增加超过默认值和/或在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

服务器代码:

"args"值:

"sendToClient"/"sendToServer"(隐含)用于传输方向.覆盖类型的"option"/"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();
            }
        }
    };
}

客户代码:

"args"值:

"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();
            }
        }
    };
}

最大的问题:

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

一个线程(netty 4.x中的ServerBootstrap.option()和ServerBootstrap.childOption()之间有什么区别)说ServerBootStrap.childOption应该用于每个客户端信息,所以我认为两个缓冲区都应该在ServerBootStrap.childOption.

Answer 1

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