博客

Netty模型

  • Post author:
  • Post category:其他

Netty模型

工作原理示意图1-简单版

Netty 主要基于主从 Reactors 多线程模型（如图）做了一定的改进，其中主从 Reactor 多线程模型有多个 Reactor
在这里插入图片描述

  1. BossGroup 线程维护Selector , 只关注Accecpt
  2. 当接收到Accept事件，获取到对应的SocketChannel, 封装成 NIOScoketChannel并注册到Worker 线程(事件循环), 并进行维护
  3. 当Worker线程监听到selector 中通道发生自己感兴趣的事件后，就进行处理(就由handler)， 注意handler 已经加入到通道

工作原理示意图2-进阶版

Netty 主要基于主从Reactors 多线程模型（如图）做了一定的改进，其中主从 Reactor 多线程模型有多个 Reactor
在这里插入图片描述

工作原理示意图-详细版

  1. Netty抽象出两组线程池 BossGroup 专门负责接收客户端的连接, WorkerGroup 专门负责网络的读写
  2. BossGroup 和 WorkerGroup 类型都是 NioEventLoopGroup
  3. NioEventLoopGroup 相当于一个事件循环组, 这个组中含有多个事件循环 ，每一个事件循环是NioEventLoop
  4. NioEventLoop 表示一个不断循环的执行处理任务的线程， 每个NioEventLoop 都有一个selector , 用于监听绑定在其上的socket的网络通讯
  5. NioEventLoopGroup 可以有多个线程, 即可以含有多个NioEventLoop
  6. 每个Boss NioEventLoop 循环执行的步骤有3步
    1. 轮询accept 事件
    2. 处理accept 事件 , 与client建立连接 , 生成NioScocketChannel , 并将其注册到某个worker NIOEventLoop 上的 selector
    3. 处理任务队列的任务 ， 即 runAllTasks
  7. 每个 Worker NIOEventLoop 循环执行的步骤
    1. 轮询read, write 事件
    2. 处理i/o事件， 即read , write 事件，在对应NioScocketChannel 处理
    3. 处理任务队列的任务 ， 即 runAllTasks
  8. 每个Worker NIOEventLoop 处理业务时，会使用pipeline(管道), pipeline

在这里插入图片描述

Netty快速入门实例-TCP服务

导包 在这里插入图片描述

客户端

package com.jhj.netty.simple;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
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 NettyClient {

    public static void main(String[] args) throws InterruptedException {
        //客户端需要一个事件循环1组
        EventLoopGroup group = new NioEventLoopGroup();

        try {
            //创建客户端启动对象
            //注意客户端使用的不是ServerBootstrap而是Bootstrap
            Bootstrap bootstrap = new Bootstrap();
            //设置相关参数
            bootstrap.group(group) //设置线程组
                    .channel(NioSocketChannel.class) //设置客户端通道的实现类（反射）
                    .handler(new ChannelInitializer<SocketChannel>() {


                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast(new NettyClientHandler());//加入自己的处理器
                        }
                    });

            System.out.println("客户端 ok..");

            //启动客户端去连接服务器端
            //关于ChannelFuture 涉及到netty的异步模型
            ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 6668).sync();

            //给关闭通道进行监听
            channelFuture.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } finally {
            group.shutdownGracefully();
        }
    }
}

handler

package com.jhj.netty.simple;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.CharsetUtil;

/**
 * 说明
 * 1.我们自定义一个Handler需要继承netty规定好的某个HandlerAdapter
 * <p>
 * 2.这时我们自定义一个Handler,才能成为一个Handler
 */
public class NettyClientHandler extends ChannelInboundHandlerAdapter {

    //当管道就绪时就会触发

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {

        System.out.println("client"+ctx);

        ctx.writeAndFlush(Unpooled.copiedBuffer("hello,server",CharsetUtil.UTF_8));
    }

    //当通道有读取事件时会触发

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ByteBuf buf = (ByteBuf) msg;
        System.out.println("服务器回复的消息："+buf.toString(CharsetUtil.UTF_8));
        System.out.println("服务器的地址"+ctx.channel().remoteAddress());
    }

    //处理异常，一般是需要关闭通道
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
    }
}

服务端

package com.jhj.netty.simple;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
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 class NettyServer {
    public static void main(String[] args) throws InterruptedException {

        //创建BossGroup和WorkerGroup

        /**
         * 说明
         * 1.创建两个线程组bossGroup和workerGroup
         * 2.bossGroup只是处理连接请求，真正的和客户端业务处理，会交给workerGroup完成
         * 3.两个都是无限循环
         * 4.bossGroup和workerGroup含有的子线程（NioEventLoop）的个数
         * //默认实际cpu核数*2
         */
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            //创建服务器端的启动对象，配置参数
            ServerBootstrap bootstrap = new ServerBootstrap();

            //使用链式编程来进行设置
            bootstrap.group(bossGroup,workerGroup) //设置两个线程组
                    .channel(NioServerSocketChannel.class) //使用NioSocketChannel 作为服务器的通道实现
                    .option(ChannelOption.SO_BACKLOG,128) //设置线程队列等待连接个数
                    .childOption(ChannelOption.SO_KEEPALIVE,true) //设置保持活动连接状态
                    .childHandler(new ChannelInitializer<SocketChannel>(){
                        //创建一个通道初始化对象（匿名对象）
                        //给pipeline设置处理器
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast(new NettyServerHandler());
                        }
                    }); //给我们的wokerGroup的EventGroup对应的通道设置处理器
            System.out.println("...服务器 is ready...");

            //绑定一个端口并且同步，生成了一个ChannelFuture 对象
            //启动服务器
            ChannelFuture cf = bootstrap.bind(6668).sync();


            //对关闭通道进行监听
            cf.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }finally {
            //优雅的关闭
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

handler

package com.jhj.netty.simple;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.CharsetUtil;

/**
 * 说明
 * 1.我们自定义一个Handler需要继承netty规定好的某个HandlerAdapter
 * <p>
 * 2.这时我们自定义一个Handler,才能成为一个Handler
 */
public class NettyServerHandler extends ChannelInboundHandlerAdapter {

    //读取数据的事件（这里我们可以读取客户端发送的消息）

    /**
     * @param ctx :上下文对象，含有管道 piepline,通道channel,地址
     * @param msg :就是客户端发送的数据，默认Object
     * @throws Exception
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {


        System.out.println("server ctx=" + ctx);

        //将msg转成一个ByteBuf
        //ByteBuf是Netty提供的，不是NIO的ByteBuffer
        ByteBuf buf = (ByteBuf) msg;

        System.out.println("客户端发送消息是：" + buf.toString(CharsetUtil.UTF_8));
        System.out.println("客户端地址" + ctx.channel().remoteAddress());
    }

    //数据读取完毕
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {

        //wirteAndFlush 是write+flush
        //将数据写入到缓存，并刷新
        //一般讲，我们对这个发送的数据进行编码
        ctx.writeAndFlush(Unpooled.copiedBuffer("hello,客户端~", CharsetUtil.UTF_8));

    }


    //处理异常，一般是需要关闭通道
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
    }
}

任务队列中的 Task 有 3 种典型使用场景

  1. 用户程序自定义的普通任务
 //读取数据的事件（这里我们可以读取客户端发送的消息）

    /**
     * @param ctx :上下文对象，含有管道 piepline,通道channel,地址
     * @param msg :就是客户端发送的数据，默认Object
     * @throws Exception
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {



        /*
        比如这里我们有一个非常耗时长的业务-》异步执行-》提交该channel对应的
        NIOEventLoop中的taskQueue中
         */
        //方法1. 用户自定义的普通任务
        // 10秒后执行 如果有多个则根据队列顺序相加 因为用一个线程
        ctx.channel().eventLoop().execute(new Runnable() {
            @Override
            public void run() {
                try {
                    Thread.sleep(10*1000);
                    ctx.writeAndFlush(Unpooled.copiedBuffer("hello",CharsetUtil.UTF_8));
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        });

        //30秒过后返回  10+20
        ctx.channel().eventLoop().execute(new Runnable() {
            @Override
            public void run() {
                try {
                    Thread.sleep(20*1000);
                    ctx.writeAndFlush(Unpooled.copiedBuffer("hello",CharsetUtil.UTF_8));
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        });



        System.out.println("server ctx=" + ctx);

        //将msg转成一个ByteBuf
        //ByteBuf是Netty提供的，不是NIO的ByteBuffer
        ByteBuf buf = (ByteBuf) msg;

        System.out.println("客户端发送消息是：" + buf.toString(CharsetUtil.UTF_8));
        System.out.println("客户端地址" + ctx.channel().remoteAddress());
    }
  1. 用户自定义定时任务
//方法2 用户自定义定时任务-》该任务提交到scheduleTaskQueue中
        //5秒后执行该定时任务
        ctx.channel().eventLoop().schedule(new Runnable() {
            @Override
            public void run() {
                try {
                    Thread.sleep(20*1000);
                    ctx.writeAndFlush(Unpooled.copiedBuffer("hello",CharsetUtil.UTF_8));
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        },5, TimeUnit.SECONDS);
  1. 非当前 Reactor 线程调用 Channel 的各种方法例如在推送系统的业务线程里面，根据用户的标识，找到对应的 Channel 引用，然后调用 Write 类方法向该用户推送消息，就会进入到这种场景。最终的 Write 会提交到任务队列中后被异步消费

方案再说明

  1. Netty 抽象出两组线程池，BossGroup 专门负责接收客户端连接，WorkerGroup 专门负责网络读写操作。
  2. NioEventLoop 表示一个不断循环执行处理任务的线程，每个 NioEventLoop 都有一个selector，用于监听绑定在其上的 socket 网络通道。
  3. NioEventLoop 内部采用串行化设计，从消息的读取->解码->处理->编码->发送，始终由IO 线程 NioEventLoop 负责
    • NioEventLoopGroup 下包含多个 NioEventLoop
    • 每个 NioEventLoop 中包含有一个 Selector，一个 taskQueue
    • 每个 NioEventLoop 的 Selector 上可以注册监听多个 NioChannel
    • 每个 NioChannel 只会绑定在唯一的 NioEventLoop 上
    • 每个 NioChannel 都绑定有一个自己的 ChannelPipelin

作者声明

如有问题，欢迎指正！