第5部分：Netty性能优化与调优策略

第5部分：Netty性能优化与调优策略

5.1 参数调优

线程数调优

1. EventLoopGroup线程数配置

1public class ThreadOptimization {
2    public void configureThreads() {
3        // 获取CPU核心数
4        int cpuCores = Runtime.getRuntime().availableProcessors();
5        
6        // Boss线程组：通常1个即可，处理连接建立
7        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
8        
9        // Worker线程组：建议CPU核心数 * 2
10        EventLoopGroup workerGroup = new NioEventLoopGroup(cpuCores * 2);
11        
12        // 业务线程组：处理耗时业务逻辑
13        EventLoopGroup businessGroup = new NioEventLoopGroup(cpuCores * 4);
14    }
15}
16

2. 线程池配置优化

1public class ThreadPoolOptimization {
2    public void configureThreadPool() {
3        // 自定义线程工厂
4        ThreadFactory threadFactory = new ThreadFactory() {
5            private final AtomicInteger counter = new AtomicInteger(0);
6            
7            @Override
8            public Thread newThread(Runnable r) {
9                Thread t = new Thread(r, "Netty-Thread-" + counter.incrementAndGet());
10                t.setDaemon(false); // 非守护线程
11                t.setPriority(Thread.NORM_PRIORITY);
12                return t;
13            }
14        };
15        
16        // 使用自定义线程工厂
17        EventLoopGroup group = new NioEventLoopGroup(8, threadFactory);
18    }
19}
20

缓冲区大小调优

1. ByteBuf分配器配置

1public class BufferOptimization {
2    public void configureBuffer() {
3        ServerBootstrap bootstrap = new ServerBootstrap();
4        
5        // 配置ByteBuf分配器
6        bootstrap.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
7        bootstrap.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
8        
9        // 配置接收缓冲区大小
10        bootstrap.option(ChannelOption.SO_RCVBUF, 64 * 1024); // 64KB
11        bootstrap.option(ChannelOption.SO_SNDBUF, 64 * 1024); // 64KB
12        
13        // 配置接收缓冲区低水位线
14        bootstrap.option(ChannelOption.RCVBUF_ALLOCATOR, 
15            new AdaptiveRecvByteBufAllocator(64, 1024, 65536));
16    }
17}
18

2. 内存分配策略

1public class MemoryAllocationStrategy {
2    public void configureMemoryAllocation() {
3        // 使用内存池分配器
4        ByteBufAllocator allocator = new PooledByteBufAllocator(
5            true,  // 优先使用直接内存
6            0,     // 页大小，0表示使用默认值
7            0,     // 最大顺序分配，0表示使用默认值
8            8192,  // 小对象阈值
9            0,     // 小对象页大小，0表示使用默认值
10            0,     // 小对象最大顺序分配，0表示使用默认值
11            true   // 使用缓存
12        );
13        
14        // 配置到Bootstrap
15        ServerBootstrap bootstrap = new ServerBootstrap();
16        bootstrap.option(ChannelOption.ALLOCATOR, allocator);
17        bootstrap.childOption(ChannelOption.ALLOCATOR, allocator);
18    }
19}
20

网络参数调优

1. TCP参数优化

1public class TcpOptimization {
2    public void configureTcpOptions() {
3        ServerBootstrap bootstrap = new ServerBootstrap();
4        
5        // 服务端选项
6        bootstrap.option(ChannelOption.SO_BACKLOG, 1024)           // 连接队列大小
7                .option(ChannelOption.SO_REUSEADDR, true)          // 地址重用
8                .option(ChannelOption.SO_KEEPALIVE, true)          // 保持连接
9                .option(ChannelOption.TCP_NODELAY, true)           // 禁用Nagle算法
10                .option(ChannelOption.SO_LINGER, 0)                // 关闭时立即返回
11                .option(ChannelOption.SO_TIMEOUT, 30000)           // 超时时间
12                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 30000); // 连接超时
13        
14        // 客户端选项
15        bootstrap.childOption(ChannelOption.SO_KEEPALIVE, true)
16                .childOption(ChannelOption.TCP_NODELAY, true)
17                .childOption(ChannelOption.SO_RCVBUF, 64 * 1024)
18                .childOption(ChannelOption.SO_SNDBUF, 64 * 1024);
19    }
20}
21

2. Linux系统参数优化

1# 修改系统参数
2echo 'net.core.somaxconn = 65535' >> /etc/sysctl.conf
3echo 'net.core.netdev_max_backlog = 5000' >> /etc/sysctl.conf
4echo 'net.ipv4.tcp_max_syn_backlog = 65535' >> /etc/sysctl.conf
5echo 'net.ipv4.tcp_fin_timeout = 30' >> /etc/sysctl.conf
6echo 'net.ipv4.tcp_keepalive_time = 1200' >> /etc/sysctl.conf
7echo 'net.ipv4.tcp_keepalive_probes = 3' >> /etc/sysctl.conf
8echo 'net.ipv4.tcp_keepalive_intvl = 15' >> /etc/sysctl.conf
9
10# 应用配置
11sysctl -p
12

5.2 零拷贝、DirectBuffer、Pooling策略

零拷贝技术

1. 文件传输零拷贝

1public class ZeroCopyExample {
2    public void fileTransfer(ChannelHandlerContext ctx, File file) throws IOException {
3        // 使用FileRegion实现零拷贝文件传输
4        FileRegion region = new DefaultFileRegion(file, 0, file.length());
5        
6        // 发送文件
7        ChannelFuture future = ctx.writeAndFlush(region);
8        
9        // 添加完成监听器
10        future.addListener(new ChannelFutureListener() {
11            @Override
12            public void operationComplete(ChannelFuture future) {
13                if (future.isSuccess()) {
14                    System.out.println("文件传输完成");
15                } else {
16                    System.err.println("文件传输失败: " + future.cause());
17                }
18            }
19        });
20    }
21}
22

2. CompositeByteBuf零拷贝

1public class CompositeByteBufExample {
2    public void compositeBuffers(ChannelHandlerContext ctx) {
3        // 创建组合缓冲区
4        CompositeByteBuf composite = ctx.alloc().compositeBuffer();
5        
6        // 添加多个ByteBuf
7        ByteBuf header = ctx.alloc().buffer(4);
8        header.writeInt(1001);
9        
10        ByteBuf body = ctx.alloc().buffer(1024);
11        body.writeBytes("Hello, Netty!".getBytes());
12        
13        // 组合缓冲区，不进行数据拷贝
14        composite.addComponents(true, header, body);
15        
16        // 发送组合缓冲区
17        ctx.writeAndFlush(composite);
18    }
19}
20

DirectBuffer优化

1. 直接内存配置

1public class DirectBufferOptimization {
2    public void configureDirectBuffer() {
3        // 配置JVM参数
4        // -XX:MaxDirectMemorySize=2g
5        // -XX:+UseDirectMemoryForDirectBuffers
6        
7        // 使用直接内存分配器
8        ByteBufAllocator allocator = new PooledByteBufAllocator(
9            true,  // 优先使用直接内存
10            0,     // 页大小
11            0,     // 最大顺序分配
12            8192,  // 小对象阈值
13            0,     // 小对象页大小
14            0,     // 小对象最大顺序分配
15            true   // 使用缓存
16        );
17        
18        // 配置到Channel
19        ServerBootstrap bootstrap = new ServerBootstrap();
20        bootstrap.option(ChannelOption.ALLOCATOR, allocator);
21        bootstrap.childOption(ChannelOption.ALLOCATOR, allocator);
22    }
23}
24

2. 直接内存监控

1public class DirectMemoryMonitor {
2    public void monitorDirectMemory() {
3        // 获取直接内存使用情况
4        MemoryMXBean memoryBean = ManagementFactory.getMemoryMXBean();
5        MemoryUsage directMemory = memoryBean.getNonHeapMemoryUsage();
6        
7        System.out.println("直接内存使用: " + directMemory.getUsed() / 1024 / 1024 + "MB");
8        System.out.println("直接内存最大: " + directMemory.getMax() / 1024 / 1024 + "MB");
9        
10        // 获取Netty内存池统计信息
11        PooledByteBufAllocator allocator = (PooledByteBufAllocator) 
12            PooledByteBufAllocator.DEFAULT;
13        
14        // 打印内存池统计信息
15        System.out.println("内存池统计: " + allocator.dumpStats());
16    }
17}
18

内存池策略

1. 内存池配置

1public class MemoryPoolConfiguration {
2    public void configureMemoryPool() {
3        // 配置内存池参数
4        PooledByteBufAllocator allocator = new PooledByteBufAllocator(
5            true,  // 优先使用直接内存
6            8192,  // 页大小
7            11,    // 最大顺序分配（2^11 = 2048）
8            256,   // 小对象阈值
9            9,     // 小对象页大小（2^9 = 512）
10            7,     // 小对象最大顺序分配（2^7 = 128）
11            true   // 使用缓存
12        );
13        
14        // 配置到Bootstrap
15        ServerBootstrap bootstrap = new ServerBootstrap();
16        bootstrap.option(ChannelOption.ALLOCATOR, allocator);
17        bootstrap.childOption(ChannelOption.ALLOCATOR, allocator);
18    }
19}
20

2. 内存池监控

1public class MemoryPoolMonitor {
2    public void monitorMemoryPool() {
3        PooledByteBufAllocator allocator = (PooledByteBufAllocator) 
4            PooledByteBufAllocator.DEFAULT;
5        
6        // 获取内存池统计信息
7        PooledByteBufAllocatorMetric metric = allocator.metric();
8        
9        System.out.println("直接内存使用: " + metric.usedDirectMemory());
10        System.out.println("堆内存使用: " + metric.usedHeapMemory());
11        System.out.println("直接内存池数量: " + metric.numDirectArenas());
12        System.out.println("堆内存池数量: " + metric.numHeapArenas());
13        
14        // 打印详细统计信息
15        System.out.println(allocator.dumpStats());
16    }
17}
18

5.3 对象复用与Recycler

对象池化

1. 使用Recycler

1public class RecyclerExample {
2    // 定义可回收对象
3    private static final Recycler<RecyclableObject> RECYCLER = new Recycler<RecyclableObject>() {
4        @Override
5        protected RecyclableObject newObject(Handle<RecyclableObject> handle) {
6            return new RecyclableObject(handle);
7        }
8    };
9    
10    public static class RecyclableObject {
11        private final Recycler.Handle<RecyclableObject> handle;
12        private String data;
13        
14        public RecyclableObject(Recycler.Handle<RecyclableObject> handle) {
15            this.handle = handle;
16        }
17        
18        public void setData(String data) {
19            this.data = data;
20        }
21        
22        public String getData() {
23            return data;
24        }
25        
26        public void recycle() {
27            // 重置状态
28            data = null;
29            // 回收到对象池
30            handle.recycle(this);
31        }
32    }
33    
34    public void useRecyclableObject() {
35        // 从对象池获取对象
36        RecyclableObject obj = RECYCLER.get();
37        
38        try {
39            // 使用对象
40            obj.setData("Hello, Recycler!");
41            System.out.println(obj.getData());
42        } finally {
43            // 回收对象
44            obj.recycle();
45        }
46    }
47}
48

2. 自定义对象池

1public class CustomObjectPool<T> {
2    private final Queue<T> pool = new ConcurrentLinkedQueue<>();
3    private final Supplier<T> factory;
4    private final Consumer<T> resetter;
5    
6    public CustomObjectPool(Supplier<T> factory, Consumer<T> resetter) {
7        this.factory = factory;
8        this.resetter = resetter;
9    }
10    
11    public T get() {
12        T obj = pool.poll();
13        if (obj == null) {
14            obj = factory.get();
15        }
16        return obj;
17    }
18    
19    public void returnObject(T obj) {
20        if (obj != null) {
21            resetter.accept(obj);
22            pool.offer(obj);
23        }
24    }
25    
26    public int size() {
27        return pool.size();
28    }
29}
30

对象复用最佳实践

1. Handler对象复用

1@ChannelHandler.Sharable
2public class ReusableHandler extends ChannelInboundHandlerAdapter {
3    // 使用@Sharable注解，表示Handler可以被多个Channel共享
4    // 注意：Handler必须是线程安全的
5    
6    @Override
7    public void channelRead(ChannelHandlerContext ctx, Object msg) {
8        // 处理消息
9        System.out.println("收到消息: " + msg);
10        
11        // 转发给下一个Handler
12        ctx.fireChannelRead(msg);
13    }
14}
15

2. 避免频繁创建对象

1public class ObjectReuseExample {
2    // 重用StringBuilder
3    private final ThreadLocal<StringBuilder> stringBuilder = 
4        ThreadLocal.withInitial(StringBuilder::new);
5    
6    // 重用ByteBuf
7    private final ThreadLocal<ByteBuf> byteBuf = 
8        ThreadLocal.withInitial(() -> Unpooled.buffer(1024));
9    
10    public void processMessage(String message) {
11        // 使用重用的StringBuilder
12        StringBuilder sb = stringBuilder.get();
13        sb.setLength(0); // 清空内容
14        sb.append("处理: ").append(message);
15        
16        // 使用重用的ByteBuf
17        ByteBuf buf = byteBuf.get();
18        buf.clear();
19        buf.writeBytes(sb.toString().getBytes());
20        
21        // 处理数据
22        // ...
23    }
24}
25

5.4 Backpressure与流量控制

背压处理

1. 写入缓冲区监控

1public class BackpressureHandler extends ChannelInboundHandlerAdapter {
2    private static final int HIGH_WATER_MARK = 64 * 1024; // 64KB
3    private static final int LOW_WATER_MARK = 32 * 1024;  // 32KB
4    
5    @Override
6    public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
7        if (ctx.channel().isWritable()) {
8            System.out.println("Channel可写，恢复发送");
9            // 恢复发送数据
10        } else {
11            System.out.println("Channel不可写，暂停发送");
12            // 暂停发送数据
13        }
14        
15        super.channelWritabilityChanged(ctx);
16    }
17    
18    @Override
19    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
20        // 检查写入缓冲区大小
21        ChannelOutboundBuffer outboundBuffer = ctx.channel().unsafe().outboundBuffer();
22        if (outboundBuffer != null) {
23            long pendingBytes = outboundBuffer.totalPendingWriteBytes();
24            
25            if (pendingBytes > HIGH_WATER_MARK) {
26                System.out.println("写入缓冲区过大，暂停读取");
27                ctx.channel().config().setAutoRead(false);
28            } else if (pendingBytes < LOW_WATER_MARK) {
29                System.out.println("写入缓冲区正常，恢复读取");
30                ctx.channel().config().setAutoRead(true);
31            }
32        }
33        
34        super.channelRead(ctx, msg);
35    }
36}
37

2. 流量控制实现

1public class FlowControlHandler extends ChannelInboundHandlerAdapter {
2    private final AtomicLong pendingWrites = new AtomicLong(0);
3    private final int maxPendingWrites = 1000;
4    
5    @Override
6    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
7        // 检查待写入数量
8        if (pendingWrites.get() >= maxPendingWrites) {
9            System.out.println("待写入数量过多，丢弃消息");
10            return;
11        }
12        
13        // 增加待写入计数
14        pendingWrites.incrementAndGet();
15        
16        // 异步写入
17        ctx.writeAndFlush(msg).addListener(new ChannelFutureListener() {
18            @Override
19            public void operationComplete(ChannelFuture future) {
20                // 减少待写入计数
21                pendingWrites.decrementAndGet();
22                
23                if (future.isSuccess()) {
24                    System.out.println("写入成功");
25                } else {
26                    System.err.println("写入失败: " + future.cause());
27                }
28            }
29        });
30    }
31}
32

流量控制策略

1. 基于时间的流量控制

1public class TimeBasedFlowControl {
2    private final RateLimiter rateLimiter;
3    
4    public TimeBasedFlowControl(double permitsPerSecond) {
5        this.rateLimiter = RateLimiter.create(permitsPerSecond);
6    }
7    
8    public boolean tryAcquire() {
9        return rateLimiter.tryAcquire();
10    }
11    
12    public void acquire() {
13        rateLimiter.acquire();
14    }
15}
16

2. 基于队列的流量控制

1public class QueueBasedFlowControl {
2    private final BlockingQueue<Object> messageQueue;
3    private final int maxQueueSize;
4    
5    public QueueBasedFlowControl(int maxQueueSize) {
6        this.maxQueueSize = maxQueueSize;
7        this.messageQueue = new LinkedBlockingQueue<>(maxQueueSize);
8    }
9    
10    public boolean offer(Object message) {
11        return messageQueue.offer(message);
12    }
13    
14    public Object poll() {
15        return messageQueue.poll();
16    }
17    
18    public int getQueueSize() {
19        return messageQueue.size();
20    }
21    
22    public boolean isFull() {
23        return messageQueue.size() >= maxQueueSize;
24    }
25}
26

5.5 Linux epoll、SO_BACKLOG、TCP_NODELAY等配置

epoll优化

1. epoll参数配置

1# 修改epoll相关参数
2echo 'net.core.somaxconn = 65535' >> /etc/sysctl.conf
3echo 'net.core.netdev_max_backlog = 5000' >> /etc/sysctl.conf
4echo 'net.ipv4.tcp_max_syn_backlog = 65535' >> /etc/sysctl.conf
5echo 'net.ipv4.tcp_fin_timeout = 30' >> /etc/sysctl.conf
6echo 'net.ipv4.tcp_keepalive_time = 1200' >> /etc/sysctl.conf
7echo 'net.ipv4.tcp_keepalive_probes = 3' >> /etc/sysctl.conf
8echo 'net.ipv4.tcp_keepalive_intvl = 15' >> /etc/sysctl.conf
9echo 'net.ipv4.tcp_tw_reuse = 1' >> /etc/sysctl.conf
10echo 'net.ipv4.tcp_tw_recycle = 1' >> /etc/sysctl.conf
11echo 'net.ipv4.tcp_congestion_control = bbr' >> /etc/sysctl.conf
12
13# 应用配置
14sysctl -p
15

2. 使用epoll传输

1public class EpollOptimization {
2    public void configureEpoll() {
3        // 使用epoll传输（仅Linux）
4        if (Epoll.isAvailable()) {
5            EventLoopGroup bossGroup = new EpollEventLoopGroup(1);
6            EventLoopGroup workerGroup = new EpollEventLoopGroup();
7            
8            ServerBootstrap bootstrap = new ServerBootstrap();
9            bootstrap.group(bossGroup, workerGroup)
10                    .channel(EpollServerSocketChannel.class)
11                    .childHandler(new ChannelInitializer<SocketChannel>() {
12                        @Override
13                        protected void initChannel(SocketChannel ch) {
14                            // 配置Handler
15                        }
16                    });
17        } else {
18            // 回退到NIO
19            EventLoopGroup bossGroup = new NioEventLoopGroup(1);
20            EventLoopGroup workerGroup = new NioEventLoopGroup();
21            
22            ServerBootstrap bootstrap = new ServerBootstrap();
23            bootstrap.group(bossGroup, workerGroup)
24                    .channel(NioServerSocketChannel.class)
25                    .childHandler(new ChannelInitializer<SocketChannel>() {
26                        @Override
27                        protected void initChannel(SocketChannel ch) {
28                            // 配置Handler
29                        }
30                    });
31        }
32    }
33}
34