在您使用 github 存储库提供的示例中，存在多个错误：
-
你直接用channelActive方法写
在netty中，有一个策略是每个handler只能同时执行一个传入的方法，这是为了让开发更容易，并且为了确保类的方法以正确的顺序执行，它也是这样做的确保方法的副作用在其他类中可见。
- 您正在打印
channelReadComplete中的消息
channelReadComplete在当前消息缓冲区清空后调用，channelRead在调用前可能会被调用多次。
构建消息或计算消息的大小是检测有多少字节进入内部的方法。对于客户端的 2 次写入可能是在没有此成帧器的情况下在服务器上读取 1 次，作为测试，我使用了 new io.netty.handler.codec.FixedLengthFrameDecoder(120)，因此我可以使用 i++ 计算到达服务器和客户端的消息量。
-
**使用重磅打印实现轻量级操作。
根据我的分析器，大部分时间都花在了调用LOG.info() 上，记录器通常是这种情况，因为它们在幕后做很多事情，比如输出流的同步。通过使记录器仅记录每 1000 条消息，我的速度得到了巨大的提升（而且由于我在双核上运行，计算机速度非常慢......）
-
大量发送代码
发送代码每次都会重新创建ByteBuf。通过重复使用ByteBuf，您可以进一步提高发送速度，您可以通过创建ByteBuf 1 次，然后在每次传递时调用.retain() 来实现。
这很容易做到：
ByteBuf buf = createMessage(MESSAGE_SIZE);
for (int i = 0; i < NUMBER_OF_MESSAGES; ++i) {
ctx.writeAndFlush(buf.retain());
}
-
减少冲洗次数
通过减少刷新次数，您可以获得更高的原生性能。对 flush() 的每次调用都是对网络堆栈的调用，以发送待处理的消息。如果我们将该规则应用于上面的代码，它将给出以下代码：
ByteBuf buf = createMessage(MESSAGE_SIZE);
for (int i = 0; i < NUMBER_OF_MESSAGES; ++i) {
ctx.write(buf.retain());
}
ctx.flush();
最终代码
有时，您只是想看看结果并亲自尝试一下：
App.java（未更改）
public class App {
public static void main( String[] args ) throws InterruptedException {
final int PORT = 8080;
runInSeparateThread(() -> new Server(PORT));
runInSeparateThread(() -> new Client(PORT));
}
private static void runInSeparateThread(Runnable runnable) {
new Thread(runnable).start();
}
}
Client.java
public class Client {
public Client(int port) {
EventLoopGroup group = new NioEventLoopGroup();
try {
ChannelFuture channelFuture = createBootstrap(group).connect("192.168.171.102", port).sync();
channelFuture.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
group.shutdownGracefully();
}
}
private Bootstrap createBootstrap(EventLoopGroup group) {
return new Bootstrap().group(group)
.channel(NioSocketChannel.class)
.option(ChannelOption.TCP_NODELAY, true)
.handler(
new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(new io.netty.handler.codec.FixedLengthFrameDecoder(200));
ch.pipeline().addLast(new ClientHandler());
}
}
);
}
}
ClientHandler.java
public class ClientHandler extends ChannelInboundHandlerAdapter {
private final Logger LOG = LoggerFactory.getLogger(ClientHandler.class.getSimpleName());
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
final int MESSAGE_SIZE = 200;
final int NUMBER_OF_MESSAGES = 200000;
new Thread(()->{
ByteBuf buf = createMessage(MESSAGE_SIZE);
for (int i = 0; i < NUMBER_OF_MESSAGES; ++i) {
ctx.writeAndFlush(buf.retain());
}}).start();
}
int i;
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
if(i++%10000==0)
LOG.info("Got a message back from the server "+(i));
((io.netty.util.ReferenceCounted)msg).release();
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
private ByteBuf createMessage(int size) {
ByteBuf message = Unpooled.buffer(size);
for (int i = 0; i < size; ++i) {
message.writeByte((byte) i);
}
return message;
}
}
Server.java
public class Server {
public Server(int port) {
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ChannelFuture channelFuture = createServerBootstrap(bossGroup, workerGroup).bind(port).sync();
channelFuture.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
private ServerBootstrap createServerBootstrap(EventLoopGroup bossGroup,
EventLoopGroup workerGroup) {
return new ServerBootstrap().group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.handler(new LoggingHandler(LogLevel.INFO))
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(new io.netty.handler.codec.FixedLengthFrameDecoder(200));
ch.pipeline().addLast(new ServerHandler());
}
});
}
}
ServerHandler.java
public class ServerHandler extends ChannelInboundHandlerAdapter {
private final Logger LOG = LoggerFactory.getLogger(ServerHandler.class.getSimpleName());
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
ctx.writeAndFlush(msg).addListener(f->{if(f.cause()!=null)LOG.info(f.cause().toString());});
if(i++%10000==0)
LOG.info("Send the message back to the client "+(i));
;
}
int i;
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
// LOG.info("Send the message back to the client "+(i++));
ctx.flush();
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
}
测试结果
我决定测试一下如果我改变登录消息的频率会发生什么，这些是测试结果：
What to print: max message latency time taken*
(always) > 20000 >10 min
i++ % 10 == 0 > 20000 >10 min
i++ % 100 == 0 16000 4 min
i++ % 1000 == 0 0-3000 51 sec
i++ % 10000 == 0 <10000 22 sec
* 应该花点时间，没有做真正的基准测试，只快速运行程序一次
这表明通过减少对日志的调用量（精度），我们可以获得更好的传输速率（速度）。