单机百万连接调优和Netty应用级别调优
阅读原文时间:2021年10月03日阅读:1

作者:Grey

原文地址:单机百万连接调优和Netty应用级别调优

本文为深度解析Netty源码的学习笔记。

准备两台Linux服务器,一个充当服务端,一个充当客户端。

服务端

  • 操作系统:CentOS 7

  • 配置:4核8G

  • IP:192.168.118.138

客户端

  • 操作系统:CentOS 7

  • 配置:4核8G

  • IP:192.168.118.139

服务端和客户端均要配置java环境,基于jdk1.8。

如何模拟百万连接

如果服务端只开一个端口,客户端连接的时候,端口号是有数量限制的(非root用户,从1024到65535,大约6w),所以服务端开启一个端口,客户端和服务端的连接最多6w个左右。

为了模拟单机百万连接,我们在服务端开启多个端口,例如8000~8100,一共100个端口,客户端还是6w的连接,但是可以连接服务端的不同端口,所以就可以模拟服务端百万连接的情况。

准备服务端程序

服务端程序的主要逻辑是:

绑定8000端口一直到8099端口,一共100个端口,每2s钟统计一下连接数。

channelActive触发的时候,连接+1, channelInactive触发的时候,连接-1

代码见:Server.java

准备客户端程序

客户端程序的主要逻辑是:

循环连接服务端的端口(从8000一直到8099)。

代码见:Client.java

准备好客户端和服务端的代码后,打包成Client.jarServer.jar并上传到客户端和服务端的/data/app目录下。打包配置参考pom.xml

服务端和客户端在/data/app下分别准备两个启动脚本,其中服务端准备的脚本为startServer.sh, 客户端准备的脚本为startClient.sh,内容如下:

startServer.sh

java -jar server.jar -Xms6.5g -Xmx6.5g -XX:NewSize=5.5g -XX:MaxNewSize=5.5g -XX:MaxDirectMemorySize=1g

startClient.sh

java -jar client.jar -Xms6.5g -Xmx6.5g -XX:NewSize=5.5g -XX:MaxNewSize=5.5g -XX:MaxDirectMemorySize=1g

脚本文件见:startServer.shstartClient.sh

先启动服务端

cd /data/app/ 

./startServer.sh

查看日志,待服务端把100个端口都绑定好以后。

在启动客户端

cd /data/app/

./startClient.sh

然后查看服务端日志,服务端在支撑了3942个端口号以后,报了如下错误:

Caused by: java.io.IOException: Too many open files
 at sun.nio.ch.FileDispatcherImpl.init(Native Method)
 at sun.nio.ch.FileDispatcherImpl.<clinit>(FileDispatcherImpl.java:35)

突破局部文件句柄限制

使用ulimit -n命令可以查看一个jvm进程最多可以打开的文件个数,这个是局部文件句柄限制,默认是1024,我们可以修改这个值

vi /etc/security/limits.conf

增加如下两行

*               hard    nofile             1000000
*               soft    nofile             1000000

以上配置表示每个进程可以打开的最大文件数是一百万。

突破全局文件句柄限制

除了突破局部文件句柄数限制,还需要突破全局文件句柄数限制,修改如下配置文件

vi /proc/sys/fs/file-max

将这个数量修改为一百万

echo 1000000 > /proc/sys/fs/file-max

通过这种方式修改的配置在重启后失效,如果要使重启也生效,需要修改如下配置

vi /etc/sysctl.conf

在文件末尾加上

fs.file-max=1000000

服务端和客户端在调整完局部文件句柄限制和全局文件句柄限制后,再次启动服务端,待端口绑定完毕后,启动客户端。

查看服务端日志,可以看到,服务端单机连接数已经达到百万级别。

.....
connections: 434703
connections: 438238
connections: 441195
connections: 444082
connections: 447596
.....
connections: 920435
connections: 920437
connections: 920439
connections: 920442
connections: 920443
connections: 920445
.....

场景

服务端接受到客户端的数据,进行一些相对耗时的操作(比如数据库查询,数据处理),然后把结果返回给客户端。

模拟耗时操作

在服务端,模拟通过sleep方法来模拟耗时操作,规则如下:

  • 90.0%情况下,处理时间为1ms

  • 95.0%情况下,处理时间为10ms

  • 99.0%情况下,处理时间为100ms

  • 99.9%情况下,处理时间为1000ms

代码如下

protected Object getResult(ByteBuf data) {
    int level = ThreadLocalRandom.current().nextInt(1, 1000);
    int time;
    if (level <= 900) {
        time = 1;
    } else if (level <= 950) {
        time = 10;
    } else if (level <= 990) {
        time = 100;
    } else {
        time = 1000;
    }
    try {
        Thread.sleep(time);
    } catch (InterruptedException e) {
    }
    return data;
}

客户端统计QPS和AVG逻辑

获取当前时间戳,客户端在和服务端建立连接后,会每隔1s给服务端发送数据,发送的数据就是当前的时间戳,服务端获取到这个时间戳以后,会把这个时间戳再次返回给客户端,所以客户端会拿到发送时候的时间戳,然后客户端用当前时间减去收到的时间戳,就是这个数据包的处理时间,记录下这个时间,然后统计数据包发送的次数,根据这两个变量,可以求出QPS和AVG,其中:

QPS 等于 总的请求量 除以 持续到当前的时间

AVG 等于 总的响应时间除以请求总数

客户端源码参考:Client.java

服务端源码参考:Server.java

服务端在不做任何优化的情况下,关键代码如下

...
bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel ch) {
                ch.pipeline().addLast(new FixedLengthFrameDecoder(Long.BYTES));
                ch.pipeline().addLast(/*businessGroup,*/ ServerBusinessHandler.INSTANCE);
//                ch.pipeline().addLast(ServerBusinessThreadPoolHandler.INSTANCE);
            }
        });
...


@ChannelHandler.Sharable
public class ServerBusinessHandler extends SimpleChannelInboundHandler<ByteBuf> {
    public static final ChannelHandler INSTANCE = new ServerBusinessHandler();

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) {
        ByteBuf data = Unpooled.directBuffer();
        data.writeBytes(msg);
        Object result = getResult(data);
        ctx.channel().writeAndFlush(result);
    }

    protected Object getResult(ByteBuf data) {
        int level = ThreadLocalRandom.current().nextInt(1, 1000);
        int time;
        if (level <= 900) {
            time = 1;
        } else if (level <= 950) {
            time = 10;
        } else if (level <= 990) {
            time = 100;
        } else {
            time = 1000;
        }

        try {
            Thread.sleep(time);
        } catch (InterruptedException e) {
        }

        return data;
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        // ignore
    }
}

运行服务端和客户端,查看客户端日志

.....
qps: 1466, avg response time: 35.68182
qps: 832, avg response time: 214.28384
qps: 932, avg response time: 352.59363
qps: 965, avg response time: 384.59448
qps: 957, avg response time: 403.33804
qps: 958, avg response time: 424.5246
qps: 966, avg response time: 433.35272
qps: 980, avg response time: 484.2116
qps: 986, avg response time: 478.5395
.....

优化方案一:使用自定义线程池处理耗时逻辑

将服务端代码做如下调整

bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel ch) {
                ch.pipeline().addLast(new FixedLengthFrameDecoder(Long.BYTES));
                //ch.pipeline().addLast(/*businessGroup,*/ ServerBusinessHandler.INSTANCE);
                ch.pipeline().addLast(ServerBusinessThreadPoolHandler.INSTANCE);
            }
        });

其中ServerBusinessThreadPoolHandler中,使用了自定义的线程池来处理耗时的getResult方法。关键代码如下:

private static ExecutorService threadPool = Executors.newFixedThreadPool(1000);
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) {
        ByteBuf data = Unpooled.directBuffer();
        data.writeBytes(msg);
        threadPool.submit(() -> {
            Object result = getResult(data);
            ctx.channel().writeAndFlush(result);
        });

    }

再次运行服务端和客户端,可以查看客户端日志,QPS和AVG指标都有明显的改善

....
qps: 1033, avg response time: 17.690498
qps: 1018, avg response time: 17.133448
qps: 1013, avg response time: 15.563113
qps: 1010, avg response time: 15.415672
qps: 1009, avg response time: 16.049961
qps: 1008, avg response time: 16.179882
qps: 1007, avg response time: 16.120466
qps: 1006, avg response time: 15.822202
qps: 1006, avg response time: 15.987518
....

实际生产过程中,Executors.newFixedThreadPool(1000);中配置的数量需要通过压测来验证。

优化方案二:使用Netty原生的线程池优化

我们可以通过Netty提供的线程池来处理耗时的Handler,这样的话,无需调整Handler的逻辑(对原有Handler无代码侵入),关键代码:

bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel ch) {
                ch.pipeline().addLast(new FixedLengthFrameDecoder(Long.BYTES));
                // ch.pipeline().addLast(ServerBusinessHandler.INSTANCE);
                // 使用业务线程池方式
                // ch.pipeline().addLast(ServerBusinessThreadPoolHandler.INSTANCE);
                // 使用Netty自带线程池方式
                ch.pipeline().addLast(businessGroup,ServerBusinessHandler.INSTANCE);
            }
        });

其中businessGroup是Netty自带的线程池

EventLoopGroup businessGroup = new NioEventLoopGroup(1000);

ServerBusinessHandler中的所有方法,都会在businessGroup中执行。

再次启动服务端和客户端,查看客户端日志

.....
qps: 1027, avg response time: 23.833092
qps: 1017, avg response time: 20.98855
qps: 1014, avg response time: 18.220013
qps: 1012, avg response time: 17.447332
qps: 1010, avg response time: 16.502508
qps: 1010, avg response time: 15.692251
qps: 1009, avg response time: 15.968423
qps: 1008, avg response time: 15.888149
.....

参考Netty性能调优奇技淫巧还有其他的吗?

1.如果QPS过高,数据传输过快的情况下,调用writeAndFlush可以考虑拆分成多次write,然后单次flush,也就是批量flush操作

2.分配和释放内存尽量在reactor线程内部做,这样内存就都可以在reactor线程内部管理

3.尽量使用堆外内存,尽量减少内存的copy操作,使用CompositeByteBuf可以将多个ByteBuf组合到一起读写

4.外部线程连续调用eventLoop的异步调用方法的时候,可以考虑把这些操作封装成一个task,提交到eventLoop,这样就不用多次跨线程

5.尽量调用ChannelHandlerContext.writeXXX()方法而不是channel.writeXXX()方法,前者可以减少pipeline的遍历

6.如果一个ChannelHandler无数据共享,那么可以搞成单例模式,标注@Shareable,节省对象开销对象

7.如果要做网络代理类似的功能,尽量复用eventLoop,可以避免跨reactor线程

Github

深度解析Netty源码