Netty UDP 性能问题

Question

我已经实现了三个小型 UDP 服务器。一个带有普通的 Java DatagramSocket（线程），一个带有 Netty，最后一个也带有 Netty，但带有线程消息处理（因为 Netty 不支持使用 UDP 的多个线程）。

经过一些测量，我得到以下每秒请求的结果：

• DatagramSocket ~30.000 个请求/秒
• Netty ~1.500 请求/秒
• Netty（线程）：~8.000 个请求/秒

我必须实现的实际应用程序必须处理 > 25.000 个请求/秒。所以我的问题是，如果我在 Netty 上出了问题，或者 Netty 的设计目的不是每秒处理那么多连接？

这里是实现

DatagramSocket 主要

public static void main(String... args) throws Exception {
    final int port = Integer.parseInt(args[0]);
    final int threads = Integer.parseInt(args[1]);
    final int work = Integer.parseInt(args[2]);

    DATAGRAM_SOCKET = new DatagramSocket(port);

    for (int i = 0; i < threads; i++) {
        new Thread(new Handler(work)).start();
    }
}

数据报套接字处理程序

private static final class Handler implements Runnable {
    private final int work;

    public Handler(int work) throws SocketException {
        this.work = work;
    }

    @Override 
    public void run() {
        try {
            while (!DATAGRAM_SOCKET.isClosed()) {
                final DatagramPacket receivePacket = new DatagramPacket(new byte[1024], 1024);
                DATAGRAM_SOCKET.receive(receivePacket);
                final InetAddress ip = receivePacket.getAddress();
                final int port = receivePacket.getPort();
                final byte[] sendData = "Hey there".getBytes();
                Thread.sleep(RANDOM.nextInt(work));
                final DatagramPacket sendPacket = new DatagramPacket(sendData, sendData.length, ip, port);
                DATAGRAM_SOCKET.send(sendPacket);
            }
        } catch (Exception e) {
            System.out.println("ERROR: " + e.getMessage());
        }
    }
}

网络主

public static void main(String[] args) throws Exception
{
    final int port = Integer.parseInt(args[0]);
    final int sleep = Integer.parseInt(args[1]);

    final Bootstrap bootstrap = new Bootstrap();
    bootstrap.group(new NioEventLoopGroup());
    bootstrap.channel(NioDatagramChannel.class);
    bootstrap.handler(new MyNettyUdpHandler(sleep));
    bootstrap.bind(port).sync().channel().closeFuture().sync();
}

Netty 处理程序（线程）

public class MyNettyUdpHandler extends MessageToMessageDecoder<DatagramPacket> {
    private final Random random = new Random(System.currentTimeMillis());
    private final int sleep;

    public MyNettyUdpHandler(int sleep) {
        this.sleep = sleep;
    }

    @Override
    protected void decode(ChannelHandlerContext channelHandlerContext, DatagramPacket datagramPacket, List list) throws Exception {
        new Thread(() -> {
            try {
                Thread.sleep(random.nextInt(sleep));
            } catch (InterruptedException e) {
                System.out.println("ERROR while sleeping");
            }

            final ByteBuf buffer = Unpooled.buffer(64);
            buffer.writeBytes("Hey there".getBytes());
            channelHandlerContext.channel().writeAndFlush(new DatagramPacket(buffer, datagramPacket.sender()));
        }).start();
    }
}

无线程的 Netty Handler 是一样的，只是没有线程。

Answer 1

在每个 decode() 中创建一个线程是低效的。 你可以按照Eran所说的提交任务到channel.eventLoop()，如果任务简单并且不会阻塞（实际上MesaggeToMessageDecoders中的decode()是由频道的EventLoop执行的，所以你不需要手动提交，除非你想摆脱它）。 或者您可以将任务提交给ThreadPoolExecutor 或EventExecutorGroup。 后者更好，因为您可以将侦听器添加到EventExecutorGroup.submit() 返回的Future，这样您就不必等待任务完成。 我的英语很差，希望对你有所帮助。

编辑： 可以这样写，只需在EventLoop（即I/O线程）中执行简单的逻辑代码：

@Override
protected void decode(ChannelHandlerContext channelHandlerContext, DatagramPacket datagramPacket, List list) throws Exception {
        //do something simple with datagramPacket
        ...

        final ByteBuf buffer = Unpooled.buffer(64);
        buffer.writeBytes("Hey there".getBytes());
        channelHandlerContext.channel().writeAndFlush(new DatagramPacket(buffer, datagramPacket.sender()));
}

Answer 2

您可以像这样更改您的 Netty decode() 方法，使其等效于 DatagramSocket 代码：

@Override
protected void decode(ChannelHandlerContext channelHandlerContext, DatagramPacket datagramPacket, List list) throws Exception {
  final Channel channel = channelHandlerContext.channel();
  channel.eventLoop().schedule(() -> {
    final ByteBuf buffer = Unpooled.buffer(64);
    buffer.writeBytes("Hey there".getBytes());
    channel.writeAndFlush(new DatagramPacket(buffer, datagramPacket.sender()));
  }, random.nextInt(sleep), TimeUnit.MILLISECONDS);
}

但我猜 sleep() 代码正在模拟您稍后将执行的业务代码。 如果是这种情况，请确保您没有在处理程序中运行阻塞代码。

编辑：

在下面回答您的问题： 你对频道有点困惑。您在引导程序中创建一个管道，并绑定到某个端口。返回的通道是服务器通道。 handlers 方法中的通道（在您的情况下是您的 decode 方法），就像您在传统套接字编程中 accept() 时获得的套接字。请注意，您从传入的 DatagramPacket 中提取的端口 - 它大致相同。因此，您通过此通道将数据发送回客户端。

我编写的用于安排响应的代码与您的 DatagramSocket 代码和您编写的线程化网络代码完全相同。 我不确定您为什么这样做，只是假设您有延迟响应的业务需求。 如果不是这种情况，您可以删除调度调用，您的代码将运行得更快。 如果您的业务逻辑是非阻塞的，并且运行在几毫秒内，那么您就完成了。如果它是阻塞的，你需要尝试找到一个非阻塞的替代方案，或者在执行器中运行它，即不在事件循环上。

希望这会有所帮助，即使这不是您最初问题的一部分。 Netty 很棒，我讨厌看到关于它的坏例子和坏情绪，所以我想这值得我花时间;)