Reactor线程模型

• Reactor线程模型不是java专属，也不是Netty专属，他其实是一种并发编程模型，是一种思想，具有指导意义。比如，Netty就是结合了NIO的特点，应用了Reactor线程模型实现的

• Reactor模型中定义了三种角色
  Reactor：负责监听和分配事件，将I/O事件分派给对应的Handler。新的事件包连续建立就绪、读就绪、写就绪等
  Acceptor：处理客户端新连接，并分派请求到处理器链中
  Handler：将自身与事件绑定，执行非阻塞读/写任务，完成channel的读入，完成处理业务逻辑后，负责将结果写出channel

• 常见的Reactor线程模型有三种

  1. Reactor单线程模型
  2. Reactor多线程模型
  3. 主从Reactor多线程模型

单线程模型

单Reactor单线程模型-优缺点

• 优点
  结构简单，有单线程完成，没有多线程，进程通信等问题
  适合用在一些业务逻辑比较简单，对于性能要求不高的应用场景
• 缺点
  由于单线程操作，不能充分发挥多核cpu的性能
  当Reactor线程负载过重后，处理速度变慢，这会导致大量客户端连接超时，超时之后往往会进行重发，这更加重Reactor线程的负载，最终导致大量消息积压和处理超时成为系统性能的瓶颈
  可靠性差，如果该线程进入死循环或意外终止，就会导致整个通信系统不可用，容易造成单点故障

单Reactor线程模型

存在的问题：
多线程数据共享和访问比较复杂。如果子线程完成业务处理后，把结果传递给主线程的Reactor进行发送，就会涉及共享数据的互斥和保护机制
Reactor承担所有的事件的监听和响应，只在主线程种运行，可能会存在性能问题。例如并发百万客户端连接，或者服务端需要对客户端握手进行安全认证，但是认证本身飞上损耗性能

主从Reactor线程模型

Netty模型

案例

使用Netty既可以编写服务端，也可以编写客户端，但是我们学习的重点是服务端的实现

下面将使用Netty实现简单的socket服务端，具体实现功能如下

• 服务端启动后，监听6666端口
• 服务端接收请求后，在控制台接收到数据打印出来
• 最后给客户端响应“ok”

maven依赖：

  <dependency>
        <groupId>io.nettygroupId>
        <artifactId>netty-allartifactId>
        <version>4.1.66.Finalversion>
    dependency>
1
2
3
4
5

public class NettyServer {
    public static void main(String[] args) {
        // 主，Reactor 定义BossGroup,用于接收用户的连接请求
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        // 从，Reactor 定义WorkerGroup，用于业务逻辑的处理，默认线程数：cpu核数*2
        EventLoopGroup workGroup = new NioEventLoopGroup();
        // 构建Netty服务的辅助类
        ServerBootstrap serverBootstrap = new ServerBootstrap();
        try{
            serverBootstrap.group(bossGroup,workGroup)
                    .channel(NioServerSocketChannel.class)// 指定通道的处理类型，使用的是NIO的方式
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            // 自定以处理器
                            ch.pipeline().addLast(new MyChannelHandler());

                        }// 业务逻辑处理
                    });
            // 启动服务，监听6666端口
            ChannelFuture future = serverBootstrap.bind(6666).sync();
            System.out.println("Netty服务器启动了");
            // 等待监听关闭的信号，阻塞当前的线程，等待客户端的请求
            future.channel().closeFuture().sync();

        }catch (Exception e){
          e.printStackTrace();
        }finally {
            bossGroup.shutdownGracefully();
            workGroup.shutdownGracefully();
        }

    }

}

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

public class MyChannelHandler extends SimpleChannelInboundHandler<ByteBuf> {
    /**
     * 接收客户端信息并处理
     * @param ctx
     * @param msg
     * @throws Exception
     */
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
        String msgStr = msg.toString(CharsetUtil.UTF_8);
        System.out.println("接收到客户端发来的消息"+msgStr);
        // 响应客户端
        ctx.writeAndFlush(Unpooled.copiedBuffer("ok",CharsetUtil.UTF_8));
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15