netty根据业务场景（IO密集型、CPU密集型）优化线程模型

业务的常用两种场景

• CPU 密集型：运算型（科学计算等，不需要依赖太多的外部资源）
• IO 密集型：等待型（调用微服务、数据库、远程服务等需要等待服务端返回）

CPU 密集型

保持当前线程模型即可：

• Runtime.getRuntime().availableProcessors() * 2：当前cpu核数*2
• io.netty.availableProcessors * 2：docker模式下设置核数*2
• io.netty.eventLoopThreads：指定线程数

IO 密集型

整改线程模型：独立出 “线程池”来处理业务，不在和NioEventLoopGroup做共享，解决IO抢占资源的问题，如果IO处理不及时就会影响业务处理效率。所以采用线程池提升业务效率

• 添加 handler 时，指定1个：
  EventExecutorGroup eventExecutorGroup = new UnorderedThreadPoolEventExecutor(10);
  pipeline.addLast(eventExecutorGroup, serverHandler)

• 在handler内部使用线程池，更推荐使用netty提供的线程池

示例代码

import com.example.netty.protostuff.UnpackFormatHandler;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioChannelOption;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.LengthFieldPrepender;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.util.concurrent.DefaultThreadFactory;
import io.netty.util.concurrent.EventExecutorGroup;
import io.netty.util.concurrent.UnorderedThreadPoolEventExecutor;
import lombok.extern.slf4j.Slf4j;

//服务端代码
@Slf4j
public class PoolServer {

    public static void main(String[] args) throws Exception {
        EventLoopGroup boss = new NioEventLoopGroup(0,new DefaultThreadFactory("boss"));
        EventLoopGroup worker = new NioEventLoopGroup(0,new DefaultThreadFactory("worker"));
        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            // 定义线程池
            EventExecutorGroup business = new UnorderedThreadPoolEventExecutor(10, new DefaultThreadFactory("business"));

            bootstrap.group(boss, worker)
                    .channel(NioServerSocketChannel.class)
                    .childOption(NioChannelOption.SO_KEEPALIVE,true)
                    .childHandler(new ChannelInitializer<SocketChannel>() {

                         @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            // 固定长度字段报文来分包
                            pipeline.addLast("unpackFormat",new UnpackFormatHandler());
                            // LengthFieldPrepender是一个编码器，主要是在响应字节数据前面添加字节长度字段
                            pipeline.addLast("lengthFieldPrepender",new LengthFieldPrepender(2));

                            pipeline.addLast("decoder", new StringDecoder());
                            pipeline.addLast("encoder", new StringEncoder());

                            pipeline.addLast(business,new PoolServerHandler());
                        }
                    });
            ChannelFuture future = bootstrap.bind(9000).sync();
            future.channel().closeFuture().sync();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            worker.shutdownGracefully();
            boss.shutdownGracefully();
        }
    }

    //服务端处理handler
    public static class PoolServerHandler extends SimpleChannelInboundHandler<String> {

        @Override
        protected void channelRead0(ChannelHandlerContext ctx, String s) throws Exception {
            // 处理业务逻辑
            Thread.sleep(3000);
            log.info("业务处理完成");
        }

        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            log.info("业务链接");
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73

import com.example.netty.protostuff.UnpackFormatHandler;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldPrepender;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;

//客户端代码
@Slf4j
public class PoolClient {

    public static void main(String[] args) throws Exception {


        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try{
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(workerGroup)
                    .channel(NioSocketChannel.class)
                    .remoteAddress(new InetSocketAddress(9000))
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();

                            // 固定长度字段报文来分包
                            pipeline.addLast("unpackFormat",new UnpackFormatHandler());
                            // LengthFieldPrepender是一个编码器，主要是在响应字节数据前面添加字节长度字段
                            pipeline.addLast("lengthFieldPrepender",new LengthFieldPrepender(2));

                            pipeline.addLast("decoder", new StringDecoder());
                            pipeline.addLast("encoder", new StringEncoder());

                            pipeline.addLast(new PoolClientHandler());
                        }
                    });

            ChannelFuture channelFuture = bootstrap.connect().sync();
            Channel channel = channelFuture.channel();
            for (int i = 0; i < 20; i++) {
                channel.writeAndFlush("我是消息："+i);
                log.info("我是消息："+i);
            }
            channelFuture.channel().closeFuture().sync();
        } finally {
            workerGroup.shutdownGracefully();
        }
    }

    static class PoolClientHandler extends SimpleChannelInboundHandler<String> {

        @Override
        protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {

        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66

为什么不使用NioEventLoopGroup代替UnorderedThreadPoolEventExecutor？

• NioEventLoopGroup并不会启用多个线程，因为addLast底层childExecutor(group)时判断这个handler是否绑定group里面唯一的executor，group.next()返回的是NioEventLoopGroup的一个子元素（一个线程），而UnorderedThreadPoolEventExecutor中返回的是它自身，所以多线程执行。