Netty-核心组件

作者: 秋风落叶黄 | 来源:发表于2020-12-10 11:38 被阅读0次

Netty是一个基于NIO的网络框架，它极大的简化并优化了TCP和UDP套接字服务器等网络编程，并且性能以及安全性等方面都具备优势，并且支持多种协议，如 FTP，SMTP，HTTP 以及各种二进制和基于文本的传统协议。

下面先看一个简单的Netty案例：

服务端代码

@Slf4j
public class NettyServer {

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

        //创建BossGroup和WorkerGroup
        //说明：
        //1. 创建两个线程组bossGroup和workerGroup
        //2. bossGroup只是处理链接请求，真正的和客户端业务处理，会交给workerGroup
        //3. 两个都是无限循环的
        //4. bossGroup和workerGroup含有子线程的（NioEventLoop）的个数，默认实际cup核数*2
        NioEventLoopGroup bossGroup = new NioEventLoopGroup(1);
        NioEventLoopGroup workerGroup = new NioEventLoopGroup(8);
        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>() {//设置workerGroup的eventLoop对应的管道处理器
                        //给pipeline设置处理器
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            //可以使用集合管理SocketChannel，在推送消息时，可以将业务加入到各个chanel对应的NIOEventLoop的taskQueue或ScheduleTaskQueue
                            log.info("客户端SocketChannel hashCode={}", socketChannel.hashCode());
                            socketChannel.pipeline().addLast(new NettyServerHandler());
                        }
                    });

            log.info("服务器 is ready...");
            //涉及netty异步模型
            ChannelFuture channelFuture = bootstrap.bind(6668).sync(); //绑定一个端口，并且同步,启动服务器

            //给channelFuture注册监听器，监控我们关系的事件
            channelFuture.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {
                    if(future.isSuccess()) {
                        log.info("监听端口6668成功");
                    }else {
                        log.info("监听端口失败");
                    }
                }
            });

            log.info("判断是否是异步操作");

            //对关闭通道进行监听
            channelFuture.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

自定义handler

/**
 * 自定义handler， 需要继承netty规定好的某个handlerAdapter
 */
@Slf4j
public class NettyServerHandler extends ChannelInboundHandlerAdapter {

    /**
     * 读取数据，客户端发送的消息
     *
     * @param ctx 上线文对象，含有管道pipeline，通道，地址
     * @param msg 客户端发送的数据，默认是Object
     * @throws Exception
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

        //将msg转换成byteBuf, netty提供的，不是NIO的byteBuffer
        ByteBuf buf = (ByteBuf) msg;
        log.info("客户端发送的消息是:{}", buf.toString(UTF_8));
        log.info("客户端地址：{}", ctx.channel().remoteAddress());
    }

    /**
     * 读取数据完毕
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        //发送数，需要进行编码
        ctx.writeAndFlush(Unpooled.copiedBuffer("hello 客户端", UTF_8));
    }

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

下面详细介绍下Netty的核心组件，都是在上面案例使用过的。

Bootstrap、 ServerBootstrap

BootStrap意思是引导，一个Netty应用通常由一个Bootstrap开始，主要作用是配置整个netty程序，串联各个组件，Netty中的Bootstrap类是客户端程序的启动引导类，ServerBootstrap是服务端启动引导类

常见的方法

 public ServerBootstrap group(EventLoopGroup parentGroup, EventLoopGroup childGroup) //用户服务端设置两个EventLoopGroup

 public B channel(Class<? extends C> channelClass) //设置一个服务器端的通道实现

 public <T> B option(ChannelOption<T> option, T value) //给ServerChannel添加配置

 public <T> ServerBootstrap childOption(ChannelOption<T> childOption, T value)//用来给接收到的通道添加配置

  public ServerBootstrap childHandler(ChannelHandler childHandler)//设置业务处理类（自定义的handler) childHandler 对应workerGroup

 public ChannelFuture bind(int inetPort) //用户服务定，设置绑定的端口号

 public B handler(ChannelHandler handler) //该handler对应BossGroup生效， 

 public ChannelFuture connect(SocketAddress remoteAddress)//用户客户端，用来连接服务器

EventLoopGroup 和其实现类 NioEventLoopGroup

EventLoopGroup 是一组 EventLoop 的抽象，Netty 为了更好的利用多核 CPU 资源，一般会有多个 EventLoop 同时工作，每个 EventLoop 维护着一个 Selector 实例。
EventLoopGroup 提供 next 接口，可以从组里面按照一定规则获取其中一个 EventLoop来处理任务。在 Netty 服务器端编程中，我们一般都需要提供两个 EventLoopGroup，例如：BossEventLoopGroup 和 WorkerEventLoopGroup。
通常一个服务端口即一个 ServerSocketChannel对应一个Selector 和一个EventLoop线程。BossEventLoop 负责接收客户端的连接并将 SocketChannel 交给 WorkerEventLoopGroup 来进行 IO 处理，如下图所示

image-20201129201931515.png

BossEventLoopGroup 通常是一个单线程的 EventLoop，EventLoop 维护着一个注册了ServerSocketChannel 的 Selector 实例BossEventLoop 不断轮询 Selector 将连接事件分离出来
通常是 OP_ACCEPT 事件，然后将接收到的 SocketChannel 交给 WorkerEventLoopGroup
WorkerEventLoopGroup 会由 next 选择其中一个 EventLoop来将这个 SocketChannel 注册到其维护的 Selector 并对其后续的 IO 事件进行处理

常用方法
- public NioEventLoopGroup()，构造方法
- public Future<?> shutdownGracefully()，断开连接，关闭线程

Channel

Netty网络通信的组件，能够用于执行网络I/O操作
通过Channel可以获得当前网络连接的通道的状态
通过Channel可以获得网络连接的配置参数（如接受缓冲区大小）
Channel提供异步的网络I/O操作（如建立连接connect()，读写 read()、write()，绑定端口 bing()），异步调用意味着任何I/O调用都将立即返回，并不保证在调用结束时，请求的I/O操作已完成
调用立即返回一个ChannelFuture实例，通过注册监听器到ChannelFuture上，可以I/O操作成功，失败或取消时回调通知调用方
支持关联I/O操作与对应的处理程序
不同协议、不同的阻塞类型的连接逗游不同的Channel类型与之对应，常见的Channel类型：

NioSocketChannel //异步客户端TCP Socket连接
NioServerSocketChannel //异步的服务端TCP Socket连接
NioDatagramChannel  //异步的UDP连接
NioSctpChannel  //异步的客户端Sctp连接
NioSctpServerChannel //异步的Sctp的服务端连接，这些通道涵盖了UDP和TCP网络IO以及文件IO

Future、channelFuture

netty中所有的IO操作都是异步的，不能立刻得到消息是否被正确处理。但是可以过一会等到它执行完成或者直接注册一个监听，具体的实现就是通过Future和ChannelFutures，他们可以注册一个监听，当操作执行成功或者失败时监听会自动触发注册的监听事件。

常见方法有

public interface ChannelFuture extends Future<Void> {
    Channel channel();//返回当前执行IO操作的通道

    ChannelFuture addListener(GenericFutureListener<? extends Future<? super Void>> var1);
     ......

    ChannelFuture sync() throws InterruptedException; //等待异步操作执行完毕
}

ChanelHandler

ChannelHandler是一个接口，处理I/O事件或拦截I/O操作，并将其转发到其ChannelPipeline(业务处理链)中的下一个处理程序
ChannelHandler本身并没有提供很多方法，因为这个接口有许多的方法需要实现，方便使用期间，可以继承它的子类
ChannelHanler及其实现类一览图

image-20201129151459934.png

Pipeline和ChannelPipeline

ChannelPipeline是一个Handler集合，它负责处理和拦截inbound或者outbound的事件和操作，相当于一个贯穿Netty的链（也可以这样理解：ChannelPipeline是保存ChannelHandler的List，用于处理或拦截Channel的入站和出站事件）
ChannelPipeline实现了一种高级形式的拦截过滤器模式，使用户可以完全控制事件的处理方式，以及Channel中各个ChannelHandler如何互相交互。
在Netty中每个Channel都有且仅有一个ChannelPipeline与之对应，它们组成关系如下：

image-20201129172333074.png
- 一个 Channel 包含了一个 ChannelPipeline，而 ChannelPipeline 中又维护了一个由 ChannelHandlerContext 组成的双向链表，并且每个 ChannelHandlerContext 中又关联着一个 ChannelHandler

入站事件和出站事件在一个双向链表中，入站事件会从链表 head 往后传递到最后一个入站的 handler，出站事件会从链表 tail 往前传递到最前一个出站的 handler，两种类型的 handler 互不干扰

常用方法

ChannelPipeline addFirst(ChannelHandler... handlers)，把一个业务处理类（handler）添加到链中的第一个位置
ChannelPipeline addLast(ChannelHandler... handlers)，把一个业务处理类（handler）添加到链中的最后一个位置

ChannelHandlerContext

保存 Channel 相关的所有上下文信息，同时关联一个 ChannelHandler 对象
即ChannelHandlerContext 中包含一个具体的事件处理器 ChannelHandler ，同时ChannelHandlerContext 中也绑定了对应的 pipeline 和 Channel 的信息，方便对 ChannelHandler进行调用.
常用方法
- ChannelFuture close()，关闭通道
- ChannelOutboundInvoker flush()，刷新
- ChannelFuture writeAndFlush(Object msg) ，将数据写到 ChannelPipeline 中当前
- ChannelHandler 的下一个 ChannelHandler 开始处理（出站）

Netty-核心组件

Bootstrap、 ServerBootstrap

EventLoopGroup 和其实现类 NioEventLoopGroup

Channel

Future、channelFuture

ChanelHandler

Pipeline和ChannelPipeline

ChannelHandlerContext

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读