【发布时间】：2021-07-14 08:59:52
【问题描述】：
我已将 Netty 提供的 echo 示例修改为乒乓基准测试，该测试计算客户端发送和接收的不同消息大小的平均延迟。我遇到的问题是数据验证。
问题如下，
- 客户端通道接收缓冲区的默认容量默认为 2048。这是一个问题，因为我想测试更大的消息大小 (4 MB)。
- 经过大量迭代，接收缓冲区的容量会变小。当消息大于 64 字节时，这会导致验证错误。
任何解决这些问题的建议或帮助都会很棒。谢谢。
这是我的代码，EchoClient：
public final class EchoClient {
static final boolean SSL = System.getProperty("ssl") != null;
static final String HOST = System.getProperty("host", "127.0.0.1");
static final int PORT = Integer.parseInt(System.getProperty("port", "8007"));
static final int SIZE = Integer.parseInt(System.getProperty("size", "256"));
static final String TRANSPORT = System.getProperty("transport", "Nio");
static final int ITER = Integer.parseInt(System.getProperty("iter", "1000"));
static final int WARMUP_ITERS = Integer.parseInt(System.getProperty("warmupIters", Integer.toString(ITER/10)));
public static void main(String[] args) throws Exception {
// Configure SSL.git
final SslContext sslCtx;
if (SSL) {
sslCtx = SslContextBuilder.forClient()
.trustManager(InsecureTrustManagerFactory.INSTANCE).build();
} else {
sslCtx = null;
}
// Configure the transport channel
Class transportChannel = (TRANSPORT.equals("Nio")) ? NioSocketChannel.class : OioSocketChannel.class;
// Configure the client.
EventLoopGroup group = (TRANSPORT.equals("Nio")) ? new NioEventLoopGroup() : new OioEventLoopGroup();
try {
Bootstrap b = new Bootstrap();
b.group(group)
.channel(transportChannel)
.option(ChannelOption.TCP_NODELAY, true)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline p = ch.pipeline();
if (sslCtx != null) {
p.addLast(sslCtx.newHandler(ch.alloc(), HOST, PORT));
}
//p.addLast(new LoggingHandler(LogLevel.INFO));
p.addLast(new EchoClientHandler());
}
});
// Start the client.
ChannelFuture f = b.connect(HOST, PORT).sync();
// Wait until the connection is closed.
f.channel().closeFuture().sync();
} finally {
// Shut down the event loop to terminate all threads.
group.shutdownGracefully();
}
}
}
回声服务器：
public final class EchoServer {
static final boolean SSL = System.getProperty("ssl") != null;
static final int PORT = Integer.parseInt(System.getProperty("port", "8007"));
static final String TRANSPORT = System.getProperty("transport", "Nio");
public static void main(String[] args) throws Exception {
// Configure SSL.
final SslContext sslCtx;
if (SSL) {
SelfSignedCertificate ssc = new SelfSignedCertificate();
sslCtx = SslContextBuilder.forServer(ssc.certificate(), ssc.privateKey()).build();
} else {
sslCtx = null;
}
// Configure the transport channel
Class transportChannel = (TRANSPORT.equals("Nio")) ? NioServerSocketChannel.class : OioServerSocketChannel.class;
// Configure the server
EventLoopGroup bossGroup = (TRANSPORT.equals("Nio")) ? new NioEventLoopGroup(1) : new OioEventLoopGroup(1);
EventLoopGroup workerGroup = (TRANSPORT.equals("Nio")) ? new NioEventLoopGroup() : new OioEventLoopGroup();
final EchoServerHandler serverHandler = new EchoServerHandler();
try {
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
.channel(transportChannel)
.option(ChannelOption.SO_BACKLOG, 100)
.handler(new LoggingHandler(LogLevel.INFO))
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline p = ch.pipeline();
if (sslCtx != null) {
p.addLast(sslCtx.newHandler(ch.alloc()));
}
//p.addLast(new LoggingHandler(LogLevel.INFO));
p.addLast(serverHandler);
}
});
// Start the server.
ChannelFuture f = b.bind(PORT).sync();
// Wait until the server socket is closed.
f.channel().closeFuture().sync();
} finally {
// Shut down all event loops to terminate all threads.
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
}
EchoClientHandler：
public class EchoClientHandler extends ChannelInboundHandlerAdapter {
long startTime;
long endTime;
long durations = 0;
private ChannelHandlerContext ctx;
int iter = EchoClient.ITER;
int skip = EchoClient.WARMUP_ITERS;
int j = iter;
int size = 1;
/**
* Creates a client-side handler.
*/
public EchoClientHandler() {
}
private ByteBuf createMsg(int size) {
ByteBuf message = Unpooled.directBuffer(size);
for (int i = 0; i < message.capacity(); i ++) {
message.writeByte((byte) i);
}
return message;
}
private void checkDataValidity(ByteBuf msgRecvByClient) {
ByteBuf msgSentByClient = createMsg(size);
if (!msgRecvByClient.equals(msgSentByClient)) {
System.err.println("ERROR: Message was corrupted");
ctx.close();
}
}
@Override
public void channelActive(ChannelHandlerContext ctx) {
this.ctx = ctx;
ByteBuf msg = createMsg(size);
startTime = System.nanoTime();
ctx.writeAndFlush(msg);
j--;
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
checkDataValidity((ByteBuf)msg);
if (j < iter - skip) {
endTime = System.nanoTime();
durations += (endTime - startTime)/(double)1000;
if (j == 0) {
System.out.printf("%d bytes has latency avg: %10.2f us\n", size, durations/((double)(iter-skip)));
durations = 0;
j = iter;
size *= 2;
if (size > EchoClient.SIZE) ctx.close();
}
}
ByteBuf newMsg = createMsg(size);
startTime = System.nanoTime();
ctx.write(newMsg);
j--;
}
@Override
public void channelReadComplete(ChannelHandlerContext ctx) {
//ByteBuf newMsg = createMsg(size);
//startTime = System.nanoTime();
//ctx.writeAndFlush(newMsg);
//j--;
ctx.flush();
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
// Close the connection when an exception is raised.
cause.printStackTrace();
ctx.close();
}
}
EchoServerHandler：
public class EchoServerHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
ctx.write(msg);
}
@Override
public void channelReadComplete(ChannelHandlerContext ctx) {
ctx.flush();
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
// Close the connection when an exception is raised.
cause.printStackTrace();
ctx.close();
}
}
【问题讨论】：
标签： netty