netty 理论知识（个人笔记）

缺乏、安全感 2022-02-01 03:34 103阅读 0赞

# #

## **Echo(helloworld)** ##

什么是Echo服务：就是一个应答服务（回显服务器），客户端发送什么数据，服务端就响应的对应的数据

1）EventLoop和EventLoopGroup  
        2)  Bootstrapt启动引导类  
        3）Channel 生命周期，状态变化  
        4）ChannelHandler和ChannelPipline

public class EchoServer {
    
        private int port;
        public EchoServer(int port){
            this.port = port;
        }
    
        /**
         * 启动流程
         */
        public void run() throws InterruptedException {
            //配置服务端线程组
            EventLoopGroup bossGroup = new NioEventLoopGroup();
            EventLoopGroup workGroup = new NioEventLoopGroup();
            try{
                ServerBootstrap serverBootstrap = new ServerBootstrap();
                serverBootstrap.group(bossGroup, workGroup)
                        //指定channel类型
                        .channel(NioServerSocketChannel.class)
                        .childHandler(new ChannelInitializer<SocketChannel>() {
    
                            protected void initChannel(SocketChannel ch) throws Exception {
                                ch.pipeline().addLast(new EchoServerHandler());
                            }
                        });
                System.out.println("Echo 服务器启动ing");
                //绑定端口，同步等待成功
                ChannelFuture channelFuture = serverBootstrap.bind(port).sync();
                //等待服务端监听端口关闭
                channelFuture.channel().closeFuture().sync();
            }finally {
                //优雅退出，释放线程池
                workGroup.shutdownGracefully();
                bossGroup.shutdownGracefully();
            }
        }
    
        public static void main(String [] args) throws InterruptedException {
            int port = 8080;
            if(args.length > 0){
                port = Integer.parseInt(args[0]);
            }
            new EchoServer(port).run();
        }
    
    }

public class EchoClient {
    
        private String host;
        private int port;
    
        public EchoClient(String host, int port){
            this.host = host;
            this.port = port;
        }
    
        public void start() throws InterruptedException {
            EventLoopGroup group = new NioEventLoopGroup();
            try {
                Bootstrap bootstrap = new Bootstrap();
                bootstrap.group(group)
                        .channel(NioSocketChannel.class)
                        .remoteAddress(new InetSocketAddress(host, port))
                        .handler(new ChannelInitializer<SocketChannel>() {
    
                            protected void initChannel(SocketChannel ch) throws Exception {
                                ch.pipeline().addLast(new EchoClientHandler());
                            }
                        });
                //连接到服务端，connect是异步连接，在调用同步等待sync，等待连接成功
                ChannelFuture channelFuture = bootstrap.connect().sync();
                //阻塞直到客户端通道关闭
                channelFuture.channel().closeFuture().sync();
            }finally {
                //优雅退出，释放NIO线程组
                group.shutdownGracefully();
            }
        }
    
        public static void main(String []args) throws InterruptedException {
            new EchoClient("127.0.0.1",8080).start();
        }
    }

public class EchoServerHandler extends ChannelInboundHandlerAdapter {
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            ByteBuf data = (ByteBuf) msg;
            System.out.println("服务端收到数据: "+ data.toString(CharsetUtil.UTF_8));
            ctx.writeAndFlush(data);
        }
    
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
            System.out.println("EchoServerHandle channelReadComplete");
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            cause.printStackTrace();
            ctx.close();
        }
    
    }

public class EchoClientHandler extends SimpleChannelInboundHandler<ByteBuf> {
    
        protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
            System.out.println("Client received: " + msg.toString(CharsetUtil.UTF_8));
        }
    
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            System.out.println("Active");
            ctx.writeAndFlush(Unpooled.copiedBuffer("课堂 xdclass.net",CharsetUtil.UTF_8));
        }
    
    
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
            System.out.println("EchoClientHandler channelReadComplete");
        }
    
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            cause.printStackTrace();
            ctx.close();
        }
    }

## **EventLoop和EventLoopGroup线程模型** ##

高性能RPC框架的3个要素：IO模型、数据协议（http、protobuf、Thrift）、线程模型

EventLoopGroup 线程组（可看作线程池）
    EventLoopGroup bossGroup = new NioEventLoopGroup();//处理新连接的线程组
    EventLoopGroup workGroup = new NioEventLoopGroup();//处理连接后收发消息的线程组

**EventLoop （可看作线程），**EventLoopGroup由多个**EventLoop组成**

每个**EventLoop**维护一个Selector（选择器），服务多个channel

NIO 种Selecto介绍：r[http://ifeve.com/selectors/][http_ifeve.com_selectors]

**Selector**

Selector（选择器）是Java NIO中能够检测一到多个NIO通道，并能够知晓通道是否为诸如读写事件做好准备的组件。这样，一个单独的线程可以管理多个channel，从而管理多个网络连接。

## **Channel（封装了的socket）** ##

Channel: 客户端和服务端建立的一个连接通道  
ChannelHandler： 负责Channel的逻辑处理  
ChannelPipeline: 负责管理ChannelHandler的有序容器

相互关系：一个Channel包含一个ChannelPipeline，所有ChannelHandler都会顺序加入到ChannelPipeline中

状态（生命周期）：

1.  channelRegistered: channel注册到一个EventLoop（注册）
2.  channelActive: 变为活跃状态（连接到了远程主机），可以接受和发送数据（启用）
3.  channelInactive: channel处于非活跃状态，没有连接到远程主机（关闭）
4.  channelUnregistered: channel已经创建，但是未注册到一个EventLoop里面，也就是没有和Selector绑定（移除）

**ChannelPipeline中添加handler**

ch.pipeline().addLast(new ReadTimeoutHandler(5)); //5秒没有相互发送消息自动断开连接

sc.pipeline().addLast("idleStateHandler", new IdleStateHandler(60, 20, 30, TimeUnit.SECONDS));//可做心跳监听

上面写法可以理解为，60秒内没有读操作执行READER_IDLE操作，20秒内没有写操作（发送消息）执行
    WRITER_IDLE（这里是关闭连接），30秒内读写任意一样没操作，执行ALL_IDLE
    
    readerIdleTime时间内Read没有处理消息
    writerIdleTime时间内没有发送消息（writeAndFlush）
    allIdleTime包括以上2种
    
    上面三种情况，任意一种达到，就会调用handler中的userEventTriggered
    
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
            System.out.println("服务器循环心跳监测发送: " + new Date());
            if (evt instanceof IdleStateEvent) {
                IdleStateEvent event = (IdleStateEvent) evt;
                if (event.state() == IdleState.WRITER_IDLE) {
                    //超时关闭
                    ctx.channel().close();
                } else if (event.state() == IdleState.READER_IDLE) {
    
                } else if (event.state() == IdleState.ALL_IDLE) {
    
                }
            }
    //		ctx.fireUserEventTriggered(evt);
        }

.childHandler(new ChannelInitializer<SocketChannel>() {
    
                            protected void initChannel(SocketChannel ch) throws Exception {
                                ch.pipeline().addLast(new EchoServerHandler());
                                ch.pipeline().addLast(new EchoServerHandler());
                                ch.pipeline().addLast(new ExceptionCaughtHandler());
                            }
                        });

**ChannelHandler**

ChannelInboundHandler：(入站)
       处理输入数据和Channel状态类型改变，
       适配器 ChannelInboundHandlerAdapter（适配器设计模式）
       常用的：SimpleChannelInboundHandler
    
    ChannelOutboundHandler：(出站)
       处理输出数据，适配器 ChannelOutboundHandlerAdapter

## **启动引导类Bootstrap** ##

1) group :设置线程组模型，Reactor线程模型对比EventLoopGroup

1）单线程 （bossGroup和workGroup数量都设置成1）  
           2）多线程  （bossGroup数量设置成1，workGroup多个EventLoop）  
           3）主从线程 （bossGroup和workGroup数量都设置成多个）

2）channel：设置channel通道类型NioServerSocketChannel

3) option: 作用于每个新建立的channel，设置channel属性

option(ChannelOption.SO\_BACKLOG，4096）//等待连接的最大数量

TPC 内核维护2个队列A,B
    三次握手，在最后一次握手之前，连接存储在A队列中，最后一次握手后把连接信息由A队列转到B队列。
    A+B的长度就是SO_BACKLOG

option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000) //连接超时时间（5秒没连接上，重试或做其他操作）

option(ChannelOption.TCP\_NODELAY，true) //true TCP连接，有数据时马上发送,不会等待足够长度。false会等待足够长度

childOption(ChannelOption.SO\_KEEPALIVE, true) //保持连接（长连接）

4）handler(new LoggingHandler(LogLevel.INFO))//日志打印

5) childHandler: 绑定通道里面的数据处理

**ChannelHandlerContext  handler中上下问信息**

ChannelHandlerContext部分方法和Channel及ChannelPipeline重合,好比调用write方法
    
    2、AbstractChannelHandlerContext类 双向链表结构，next/prev分别是后继节点，和前驱节点
                        
    3、DefaultChannelHandlerContext 是实现类，但是大部分都是父类那边完成，这个只是简单的实现一些方法
    主要就是判断Handler的类型
    
    4、ChannelInboundHandler之间的传递，主要通过调用ctx里面的FireXXX()方法来实现下个handler的调用

多个Handler执行顺序

ch.pipeline().addLast(new OutboundHandler1());  
     ch.pipeline().addLast(new OutboundHandler2());  
     ch.pipeline().addLast(new InboundHandler1());  
     ch.pipeline().addLast(new InboundHandler2());

执行顺序是：InboundHandler1(channelRead)、InboundHandler2(channelRead)、OutboundHandler2(write)、OutboundHandler1(write)

1.  InboundHandler顺序执行，OutboundHandler逆序执行
2.  InboundHandler之间传递数据，通过ctx.fireChannelRead(msg)
3.  InboundHandler通过ctx.write(msg)，则会传递到outboundHandler
4.  使用ctx.write(msg)传递消息，Inbound需要放在结尾，不然outboundhandler会不执行；
5.  但是使用channel.write(msg)、pipline.write(msg)情况会不一致，都会执行
6.  outBound和Inbound谁先执行，针对客户端和服务端而言，客户端是发起请求再接受数据，先outbound再inbound，服务端则相反

Netty中的所有I/O操作都是异步的,这意味着任何I/O调用都会立即返回，而ChannelFuture会提供有关的信息I/O操作的结果或状态。

**ChannelFuture**  
     ChannelFuture状态：  
            未完成：当I/O操作开始时，将创建一个新的对象，新的最初是未完成的 - 它既没有失败，也没有成功，也没有被取消，因为I/O操作尚未完成。

已完成：当I/O操作完成，不管是成功、失败还是取消，Future都是标记为已完成的, 失败的时候也有具体的信息，例如原因失败，但请注意，即使失败和取消属于完成状态。  
            注意：不能在channelHandler中调用sync或者await方法

ChannelPromise：继承于ChannelFuture，进一步拓展用于设置IO操作的结果

## **编解码** ##

**TCP粘包拆包**

**        1）TCP拆包: 一个完整的包可能会被TCP拆分为多个包进行发送  
        2）TCP粘包: 把多个小的包封装成一个大的数据包发送, client发送的若干数据包 Server接收时粘成一包**

**        发送方的原因：TCP默认会使用Nagle算法  
        接收方的原因: TCP接收到数据放置缓存中，应用程序从缓存中读取 **

最开始接触的编码码：java序列化/反序列化（就是编解码）、url编码、base64编解码
    
    为啥jdk有编解码，还要netty自己开发编解码？
                    java自带序列化的缺点
                        1）无法跨语言
                        2) 序列化后的码流太大，也就是数据包太大
                        3) 序列化和反序列化性能比较差

其他常用编码框架： google的 protobuf（PB)、Facebook的Trift、Jboss的Marshalling、Kyro等

解码器：负责处理“入站 InboundHandler”数据  
编码器：负责“出站 OutboundHandler” 数据

### **Decoder:解码器** ###

1.  ByteToMessageDecoder  用于将字节转为消息，需要检查缓冲区是否有足够的字节
2.  ReplayingDecoder   继承ByteToMessageDecoder，不需要检查缓冲区是否有足够的字节，但是ReplayingDecoder速度略满于ByteToMessageDecoder，不是所有的ByteBuf都支持。选择：项目复杂性高则使用ReplayingDecoder，否则使用 ByteToMessageDecoder 
3.  MessageToMessageDecoder  用于从一种消息解码为另外一种消息（例如POJO到POJO）
    
             DelimiterBasedFrameDecoder： 指定消息分隔符的解码器
             LineBasedFrameDecoder: 以换行符为结束标志的解码器
             FixedLengthFrameDecoder：固定长度解码器
             LengthFieldBasedFrameDecoder：message = header+body, 基于长度解码的通用解码器
             StringDecoder：文本解码器，将接收到的对象转化为字符串，一般会与上面的进行配合，然后在后面添加业务handle

### **Encoder:编码器** ###

1.  MessageToByteEncoder   消息转为字节数组,调用write方法，会先判断当前编码器是否支持需要发送的消息类型，如果不支持，则透传；
2.  MessageToMessageEncoder   用于从一种消息编码为另外一种消息（例如POJO到POJO）

### **Codec:编解码器** ###

讲解组合编解码器类Codec

1.  ByteToMessageCodec
2.  MessageToMessageCodec

优点：成对出现，编解码都是在一个类里面完成

缺点：耦合在一起，拓展性不佳

# #

## **ByteBuf** ##

**        JDK ByteBuffer  
            共用读写索引，每次读写操作都需要Flip()  
            扩容麻烦，而且扩容后容易造成浪费**

**        Netty ByteBuf  
            读写使用不同的索引，所以操作便捷  
            自动扩容，使用便捷**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1NDE4NTE4_size_16_color_FFFFFF_t_70][]

ByteBuf：传递字节数据的容器

**ByteBuf的创建方法**  
      1）ByteBufAllocator  
              池化(Netty4.x版本后默认使用 PooledByteBufAllocator提高性能并且最大程度减少内存碎片

非池化UnpooledByteBufAllocator： 每次返回新的实例

2）Unpooled: 提供静态方法创建未池化的ByteBuf，可以创建堆内存和直接内存缓冲区

**ByteBuf使用模式**  
      堆缓存区HEAP BUFFER:  
                    优点：存储在JVM的堆空间中，可以快速的分配和释放  
                    缺点：每次使用前会拷贝到直接缓存区(也叫堆外内存)

直接缓存区DIRECR BUFFER:  
                    优点：存储在堆外内存上，堆外分配的直接内存，不会占用堆空间  
                    缺点：内存的分配和释放，比在堆缓冲区更复杂

复合缓冲区COMPOSITE BUFFER:  
                    可以创建多个不同的ByteBuf，然后放在一起，但是只是一个视图

选择：大量IO数据读写，用“直接缓存区”； 业务消息编解码用“堆缓存区”

**Nagle算法(禁用)**

简单来说：发送的消息很小的时候，同时只允许发送一条消息，并且要等收到这消息的返回确认，才能发送第二条消息。

（1）如果包长度达到MSS，则允许发送；

（2）如果该包含有FIN，则允许发送；

（3）设置了TCP\_NODELAY选项，则允许发送；

（4）未设置TCP\_CORK选项时，若所有发出去的小数据包（包长度小于MSS）均被确认，则允许发送；

（5）上述条件都未满足，但发生了超时（一般为200ms），则立即发送。

个人练习代码，按分隔符和字符长度分隔发送消息，pb编解码发送消息

[https://github.com/grakCat/netty\_stude.git][https_github.com_grakCat_netty_stude.git]

[http_ifeve.com_selectors]: http://ifeve.com/selectors/
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1NDE4NTE4_size_16_color_FFFFFF_t_70]: /images/20220201/f9737864d74d4156aecde624698ab858.png
[https_github.com_grakCat_netty_stude.git]: https://github.com/grakCat/netty_stude.git