Echo Server 与 1000 个客户端并发（丢失消息 + 错误连接）答案

【问题标题】：Echo Server concurrently with 1000 clients (lost messages + error connection)Echo Server 与 1000 个客户端并发（丢失消息 + 错误连接）
【发布时间】：2015-10-31 03:10:57
【问题描述】：

我正在阅读“Netty In Action V5”。在阅读第 2.3 和 2.4 章时，我尝试了 EchoServer 和 EchoClient 示例，当我测试一个连接到服务器的客户端时，一切正常……然后我将示例修改为多个客户端可以连接到服务器。我的目的是运行压力测试：1000 个客户端将连接到服务器，每个客户端将向服务器回显 100 条消息，当所有客户端完成后，我将获得所有进程的总时间。服务器部署在linux机器（VPS）上，客户端部署在window机器上。

在运行压力测试时，我遇到了两个问题：

一些客户收到错误信息：

java.io.IOException: An existing connection was forcibly closed by the remote host 
    at sun.nio.ch.SocketDispatcher.read0(Native Method)
    at sun.nio.ch.SocketDispatcher.read(SocketDispatcher.java:43)
    at sun.nio.ch.IOUtil.readIntoNativeBuffer(IOUtil.java:223)
    at sun.nio.ch.IOUtil.read(IOUtil.java:192)
    at sun.nio.ch.SocketChannelImpl.read(SocketChannelImpl.java:379)
    at io.netty.buffer.UnpooledUnsafeDirectByteBuf.setBytes(UnpooledUnsafeDirectByteBuf.java:447)
    at io.netty.buffer.AbstractByteBuf.writeBytes(AbstractByteBuf.java:881)
    at io.netty.channel.socket.nio.NioSocketChannel.doReadBytes(NioSocketChannel.java:242)
    at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:119)
    at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:511)
    at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:468)
    at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:382)\at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:354)
    at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:110)
    at io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:137)
    at java.lang.Thread.run(Thread.java:745)

但是有些客户端没有收到来自服务器的消息

工作环境：

Netty-all-4.0.30.Final
JDK1.8.0_25
Echo 客户端部署在 Window 7 Ultimate 上
Echo Server 部署在 Linux Centos 6 上

类 NettyClient：

public class NettyClient {
    private Bootstrap bootstrap;
    private EventLoopGroup group;

    public NettyClient(final ChannelInboundHandlerAdapter handler) {
        group = new NioEventLoopGroup();
        bootstrap = new Bootstrap();
        bootstrap.group(group);
        bootstrap.channel(NioSocketChannel.class);
        bootstrap.handler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel channel) throws Exception {
                channel.pipeline().addLast(handler);
            }
        });
    }

    public void start(String host, int port) throws Exception {
        bootstrap.remoteAddress(new InetSocketAddress(host, port));
        bootstrap.connect();
    }

    public void stop() {
        try {
            group.shutdownGracefully().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

类 NettyServer：

public class NettyServer {
    private EventLoopGroup parentGroup;
    private EventLoopGroup childGroup;
    private ServerBootstrap boopstrap;

    public NettyServer(final ChannelInboundHandlerAdapter handler) {
        parentGroup = new NioEventLoopGroup(300);
        childGroup = new NioEventLoopGroup(300);
        boopstrap = new ServerBootstrap();
        boopstrap.group(parentGroup, childGroup);
        boopstrap.channel(NioServerSocketChannel.class);
        boopstrap.childHandler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel channel) throws Exception {
                channel.pipeline().addLast(handler);
            }
        });
    }

    public void start(int port) throws Exception {
        boopstrap.localAddress(new InetSocketAddress(port));
        ChannelFuture future = boopstrap.bind().sync();
        System.err.println("Start Netty server on port " + port);
        future.channel().closeFuture().sync();
    }

    public void stop() throws Exception {
        parentGroup.shutdownGracefully().sync();
        childGroup.shutdownGracefully().sync();
    }
}

类 EchoClient

public class EchoClient {
    private static final String HOST = "203.12.37.22";
    private static final int PORT = 3344;
    private static final int NUMBER_CONNECTION = 1000;
    private static final int NUMBER_ECHO = 10;
    private static CountDownLatch counter = new CountDownLatch(NUMBER_CONNECTION);

    public static void main(String[] args) throws Exception {
        List<NettyClient> listClients = Collections.synchronizedList(new ArrayList<NettyClient>());
        for (int i = 0; i < NUMBER_CONNECTION; i++) {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        NettyClient client = new NettyClient(new EchoClientHandler(NUMBER_ECHO) {
                            @Override
                            protected void onFinishEcho() {
                                counter.countDown();
                                System.err.println((NUMBER_CONNECTION - counter.getCount()) + "/" + NUMBER_CONNECTION);
                            }
                        });
                        client.start(HOST, PORT);
                        listClients.add(client);
                    } catch (Exception ex) {
                        ex.printStackTrace();
                    }
                }
            }).start();
        }

        long t1 = System.currentTimeMillis();
        counter.await();
        long t2 = System.currentTimeMillis();
        System.err.println("Totla time: " + (t2 - t1));

        for (NettyClient client : listClients) {
            client.stop();
        }
    }

    private static class EchoClientHandler extends SimpleChannelInboundHandler<ByteBuf> {

        private static final String ECHO_MSG = "Echo Echo Echo Echo Echo Echo Echo Echo Echo Echo Echo Echo Echo Echo Echo Echo Echo Echo Echo Echo";
        private int numberEcho;
        private int curNumberEcho = 0;

        public EchoClientHandler(int numberEcho) {
            this.numberEcho = numberEcho;
        }

        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            ctx.writeAndFlush(Unpooled.copiedBuffer(ECHO_MSG, CharsetUtil.UTF_8));
        }

        @Override
        protected void channelRead0(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
            curNumberEcho++;
            if (curNumberEcho >= numberEcho) {
                onFinishEcho();
            } else {
                ctx.writeAndFlush(Unpooled.copiedBuffer(ECHO_MSG, CharsetUtil.UTF_8));
            }
        }

        protected void onFinishEcho() {

        }

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

类 EchoServer：

public class EchoServer {
    private static final int PORT = 3344;

    public static void main(String[] args) throws Exception {
        NettyServer server = new NettyServer(new EchoServerHandler());
        server.start(PORT);
        System.err.println("Start server on port " + PORT);
    }

    @Sharable
    private static class EchoServerHandler extends ChannelInboundHandlerAdapter {
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            ctx.write(msg);
        }

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

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

【问题讨论】：

由于我无法访问这些示例，您能否打印代码，特别是您如何管理管道以及每个客户端 100 次迭代后的关闭？我自己直接参考了来自 Netty github 的示例 Echo，它是一次为 1 个客户端制作的......
搜索出现了这个问题：stackoverflow.com/questions/2582036/…。您的某些客户可能发送了格式错误的数据并触发了极端情况。
如果回显处理程序看起来像 Netty 中的处理程序，那么数据的格式并不重要。但是资源可能已经耗尽，所以我要求管道配置的原因......
我已经更新了我的例子，帮我看看，非常感谢

标签： java network-programming client server netty

【解决方案1】：

你可能会改变两件事：

只创建一个客户端引导程序并为所有客户端重复使用它，而不是为每个客户端创建一个。因此，从客户端部分中提取您的引导程序构建，并只保留您在开始时所做的连接。这将限制内部线程的数量。
在达到乒乓次数时关闭客户端连接。目前您只调用空方法 onFinishEcho，这会导致客户端根本没有关闭，因此没有客户端停止...因此也没有通道关闭...

您可能对客户端的线程数有一些限制。

另外一个元素也可能是一个问题：您没有指定任何编解码器（字符串编解码器或其他），这可能导致客户端或服务器的部分发送被视为完整响应。 p>

例如，您可能有第一个“Echo Echo Echo”块发送一个包含缓冲区开头的数据包，而其他部分（更多“Echo”）将通过以后的数据包发送。

为防止这种情况发生，您应该使用一个编解码器来确保您的最终处理程序收到的是真正的完整消息，而不是部分消息。如果没有，您可能会遇到其他问题，例如服务器端尝试发送额外数据包时出错，而客户端会按预期更快地关闭通道...

【讨论】：