Netty 并行处理程序处理

Question

按照其他地方的建议，我正在尝试在 Netty 管道中并行化我的最终入站处理程序

public final class EchoServer {
    private EventLoopGroup group = new NioEventLoopGroup();
    private UnorderedThreadPoolEventExecutor workers = new UnorderedThreadPoolEventExecutor(10);

    public void start(int port) throws InterruptedException {
        try {
            Bootstrap b = new Bootstrap();
            b.group(group).channel(NioDatagramChannel.class).option(ChannelOption.SO_BROADCAST, true)
                    .handler(new ChannelInitializer<NioDatagramChannel>() {
                        @Override
                        protected void initChannel(NioDatagramChannel channel) throws Exception {
                            channel.pipeline().addLast(workers, new SimpleChannelInboundHandler<DatagramPacket>() {
                                @Override
                                public void channelRead0(ChannelHandlerContext ctx, DatagramPacket packet) throws Exception {
                                    System.err.println(packet);
                                    // Simulated database delay that I have to wait to occur before repsonding
                                    Thread.sleep(1000);
                                    ctx.write(new DatagramPacket(Unpooled.copiedBuffer("goodbye", StandardCharsets.ISO_8859_1), packet.sender()));
                                }

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

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

            b.bind(port).sync().channel().closeFuture().await();
        } finally {
            group.shutdownGracefully();
        }
    }

    public void stop() {
        group.shutdownGracefully();
    }
}

作为测试，我有十个同时连接的客户端，我正在测量处理所有请求的执行时间。正如预期的那样，1 秒的延迟和顺序执行只需要 10 多秒。我正在尝试将执行时间缩短到 2 秒以下以证明并行处理。

据我了解，将处理程序添加到具有显式分配的执行程序的管道应该并行化处理程序在执行程序中跨线程工作。

我发现在我添加并行处理时我的客户没有收到响应，而不是看到性能提高。线程睡眠用于模拟将传入数据写入数据库所需的潜在时间。我在这里做错了什么吗？

Answer 1

我解决了 Netty 对使用标准 Java 并发机制并行执行最终 UDP 处理的明显缺乏支持的问题。

public final class EchoServer {
    private EventLoopGroup group = new NioEventLoopGroup();
    private ExecutorService executors = Executors.newFixedThreadPool(10);

    public void start(int port) throws InterruptedException {
        try {
            Bootstrap b = new Bootstrap();
            b.group(group).channel(NioDatagramChannel.class).handler(new ChannelInitializer<NioDatagramChannel>() {
                @Override
                protected void initChannel(NioDatagramChannel channel) throws Exception {
                    channel.pipeline().addLast(new SimpleChannelInboundHandler<DatagramPacket>() {
                        @Override
                        public void channelRead0(ChannelHandlerContext ctx, DatagramPacket packet) throws Exception {
                            CompletableFuture.runAsync(() -> {
                                System.err.println(packet);
                                try {
                                    Thread.sleep(1000);
                                } catch (InterruptedException e) {
                                    e.printStackTrace();
                                }
                                ctx.writeAndFlush(new DatagramPacket(Unpooled.copiedBuffer("goodbye", StandardCharsets.ISO_8859_1),
                                        packet.sender()));
                            }, executors);
                        }

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

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

            b.bind(port).sync().channel().closeFuture().await(); 
        } finally {
            group.shutdownGracefully();
        }
    }

    public void stop() {
        group.shutdownGracefully();
    }
}