Netty Parallel Handler 处理

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

Netty Parallel Handler 处理

Netty Parallel Handler Processing

parallel-processing

blocking

netty