Netty 模拟百万连接

  我们知道单机的端口最多65536，除去系统使用的端口， 留给程序使用的也就6万个端口， 在需要对单机做长连接压力测试的时候，如果要测60W的长连接并发，就得找10台机器，而一般情况下我们并没有这么多的空闲机器去做这种规模的测试，那如何用两台机器模拟百万连接呢？对于TCP的连接，系统用一个4四元组来唯一标识：{server ip, server port,client ip,client port}。这里有两个变量是固定的， server ip与clinet ip。能做文章的也就是两台服务器的端口号了。如果server port 只开启一个端口的话， 那一台client最多也就 6W个连接能连上，多了因为端口的限制无法创建新的连接。如果server端多开几个端口，根据TCP的唯一标识，我们便能够模拟超过6W的连接测试了。处面是具体的代码，项目依赖netty，版本为4.1.25.Final。

• server端代码如下：

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;


public final class Server {

    static final int BEGIN_PORT = 10000;
    static final int N_PORT = 100;

    public static void main(String[] args) {
        new Server().start(BEGIN_PORT, N_PORT);
    }

    public void start(int beginPort, int nPort) {
        System.out.println("server starting....");

        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        ServerBootstrap bootstrap = new ServerBootstrap();
        bootstrap.group(bossGroup, workerGroup);
        bootstrap.channel(NioServerSocketChannel.class);
        bootstrap.childOption(ChannelOption.SO_REUSEADDR, true);

        bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel ch) throws Exception {
                                //每个连接都有个ConnectionCountHandler对连接记数进行增加
                ch.pipeline().addLast(new ConnectionCountHandler());
            }
        });

        //这里开启 10000到100099这100个端口
        for (int i = 0; i < nPort; i++) {
            int port = beginPort + i;
            bootstrap.bind(port).addListener((ChannelFutureListener) future -> {
                System.out.println("bind success in port: " + port);
            });
        }
        System.out.println("server started!");
    }
}

• 统计服务端连接数的代码

import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

public class ConnectionCountHandler extends ChannelInboundHandlerAdapter {

    //这里用来对连接数进行记数,每两秒输出到控制台
    private static final AtomicInteger nConnection = new AtomicInteger();

    static {
        Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(() -> {
            System.out.println("connections: " + nConnection.get());
        }, 0, 2, TimeUnit.SECONDS);
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        nConnection.incrementAndGet();
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) {
        nConnection.decrementAndGet();
    }

}

• client端代码

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;

public class Client {

    //服务端的IP
    private static final String SERVER_HOST = "10.200.10.146";

    static final int BEGIN_PORT = 10000;
    static final int N_PORT = 100;

    public static void main(String[] args) {
        new Client().start(BEGIN_PORT, N_PORT);
    }

    public void start(final int beginPort, int nPort) {
        System.out.println("client starting....");
        EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
        final Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(eventLoopGroup);
        bootstrap.channel(NioSocketChannel.class);
        bootstrap.option(ChannelOption.SO_REUSEADDR, true);
        bootstrap.handler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel ch) {
            }
        });

        int index = 0;
        int port;
        //从10000的端口开始，按端口递增的方式进行连接
        while (!Thread.interrupted()) {
            port = beginPort + index;
            try {
                ChannelFuture channelFuture = bootstrap.connect(SERVER_HOST, port);
                channelFuture.addListener((ChannelFutureListener) future -> {
                    if (!future.isSuccess()) {
                        System.out.println("connect failed, exit!");
                        System.exit(0);
                    }
                });
                channelFuture.get();
            } catch (Exception e) {
            }

            if (++index == nPort) {
                index = 0;
            }
        }
    }
}

在linux机器上测试的时候，如果报了Could not initialize class sun.nio.ch.FileDispatcherImpl，这是因为系统为每个TCP连接都要创建一个socket句柄，每个socket句柄同时也是一个文件句柄。解决办法如下：

• 修改linux系统进程打开文件数的限制，文件在/etc/security/limits.conf。需要增加的行如下：

# * 表示对于任何用户， hard表示硬件的限制 ， soft表示软件限制 nofile表示打开文件数
* hard nofile 1000000
* soft nofile 1000000

• 修改linux系统能够打开文件的最大限制，文件在： /proc/sys/fs/file-max。需要增加的行如下：

1000000