Netty框架的高级概念

Posted 恒哥~Bingo

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Netty框架的高级概念相关的知识,希望对你有一定的参考价值。

编码解码

回顾Netty的几个组件

  • Channel

    服务器和客户端建立的连接通道

  • ChannelPipeline

    管道,一个通道包含一个管道,管道包含一个处理器链

  • ChannelHandler

    管道中的处理器链包含多个处理器,每个处理器可以处理不同的IO事件,是双向链表结构,包含head头部和tail尾部。

    处理器分为:

    • ChannelInboundHandler

      入站消息处理器(处理进入的消息)

    • ChannelOutboundHandler

      出站消息处理器(处理出去的消息)

编码和解码

Netty在接收和发送消息时都需要进行数据转换。接收消息时(入站)需要把字节转换为自己的格式(字符串、对象等)这就是解码。发送消息时把自己的格式转换为字节,就是编码。

编码器继承MessageToMessageEncoder类 ,实现了ChannelOutboundHandler 接口

解码器继承MessageToMessageDecoder类,实现了ChannelInboundHandler接口

Netty自带的编码解码器有:

  • StringEncoder字符串编码器
  • StringDecoder字符串解码器
  • ObjectEncoder对象编码器
  • ObjectDecoder对象解码器
 bootstrap.group(bossGroup, workerGroup)
                    .channel(NioserverSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 1024)
                    .childHandler(new ChannelInitializer<SocketChannel>() 
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception 
                            ChannelPipeline pipeline = ch.pipeline();
                            //向pipeline加入字符串解码器
                            pipeline.addLast("decoder", new StringDecoder());
                            //向pipeline加入字符串编码器
                            pipeline.addLast("encoder", new StringEncoder());
                            //加入自己的业务处理handler
                            pipeline.addLast(new ChatServerHandler());
                        
                    );

以聊天室程序为例,先在管道中加入了字符串解码器,这样就能将字节转换为字符串,读取打印出来;然后加入了字符串编码器,这些在发送数据时,将字符串转换为字节发送出去。

自定义编码和解码器

开发者可以自定义编码解码器

自定义解码器

/**
 * byte转long解码器
 */
public class ByteToLongDecoder extends ByteToMessageDecoder 

	//重写编码方法
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception 
        System.out.println("解码方法被调用");
        //读取长整型需要的8个字节
        if(in.readableBytes() >= 8) 
            out.add(in.readLong());
        
    


自定义解码器

/**
 * long转byte编码器
 */
public class LongToByteEncoder extends MessageToByteEncoder<Long> 

    //重写编码方法
    @Override
    protected void encode(ChannelHandlerContext ctx, Long msg, ByteBuf out) throws Exception 
        System.out.println("编码方法被调用 " + msg);
        out.writeLong(msg);
    

服务器和客户端在添加处理器时,可以加入编码器和解码器

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

        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try 
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() 
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception 
                            ChannelPipeline pipeline = ch.pipeline();
                            //给服务器添加了long类型的解码器和编码器
                            pipeline.addLast(new ByteToLongDecoder());
                            pipeline.addLast(new LongToByteEncoder());
                            pipeline.addLast(new NettyServerHandler());
                        
                    );

            System.out.println("netty server start。。");
            ChannelFuture channelFuture = serverBootstrap.bind(9000).sync();
            channelFuture.channel().closeFuture().sync();
         finally 
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        
    

服务器和客户端的业务处理器中,就可以读取和发送long的数据

public class NettyServerHandler extends ChannelInboundHandlerAdapter 

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception 
        System.out.println("从客户端读取到Long:" + (Long)msg);
        //给客户端发回一个long数据
        ctx.writeAndFlush(2000L);
    

粘包拆包

Netty是基于TCP协议的,TCP是面向流的,数据没有边界,操作系统在发送时会通过缓冲区进行优化,缓冲区有固定大小,如1024字节。

粘包

​ 如果一次发送数据时,数据量没有达到缓冲区大小,TCP会将多个数据包合到一起发送,这样就形成了粘包

拆包

​ 如果一次发送数据时,数据量超过了缓冲区大小,TCP会将数据拆分成多个数据包发送,这样就是拆包

如下图所示,出现几种情况

  1. 理想情况,两次发送的数据包正好是缓冲区大小,就正常作为两个数据包发送
  2. 出现粘包,两次发送的数据包加一起等于缓冲区大小,将两个包合并为一个包发送
  3. 出现拆包,一次发送的包超过了缓冲区,分成了多个包发送
  4. 拆包/粘包同时发生,一个包比缓冲区大,一个比缓冲区小,会将大的包拆开和小的包合并到一起发送


数据一般有完整性,如:文件、字符串等,发生粘包和拆包会导致数据出现混乱。

解决方案:

  1. 规定固定长度,如100字节,按长度拆分
  2. 设置某些字符作为分隔符,如:\\r\\n,读取后按分隔符进行数据拆分
  3. 数据包中包含长度,读取到后对数据进行拆分
  4. 使用特定的协议,消息头中包含长度,进行粘包和拆包处理

Netty提供了一些解码器(Decoder)来解决粘包和拆包的问题。如:

  • LineBasedFrameDecoder 以行(\\r\\n)为单位进行数据包的解码;
  • DelimiterBasedFrameDecoder:以特殊的符号作为分隔来进行数据包的解码;
  • FixedLengthFrameDecoder:以固定长度进行数据包的解码;
  • LengthFieldBasedFrameDecoder:适用于消息头包含消息长度的协议;

零拷贝

Java的内存按分配位置分为:

  • 堆内内存(JVM堆中分配的内存)
  • 直接内存(JVM外部分配的内存)

Socket网络通信时,数据通过网络发送到网卡,先通过网卡保存到直接内存中,再从直接内存中把数据复制到JVM堆中,才能对数据进行操作。


Netty采用的是零拷贝机制,在JVM中直接引用直接内存中的数据,不需要进行拷贝,提高了IO效率。

ByteBuffer类的allocate方法用于分配堆内内存,allocateDirect方法用于分配直接内存

/**
 * 直接内存与堆内存的区别
 */
public class DirectMemoryTest 

    /**
     * 访问堆内存
     */
    public static void heapAccess() 
        long startTime = System.currentTimeMillis();
        //分配堆内存
        ByteBuffer buffer = ByteBuffer.allocate(1000);
        for (int i = 0; i < 100000; i++) 
            for (int j = 0; j < 200; j++) 
                buffer.putInt(j);
            
            buffer.flip();
            for (int j = 0; j < 200; j++) 
                buffer.getInt();
            
            buffer.clear();
        
        long endTime = System.currentTimeMillis();
        System.out.println("堆内存访问:" + (endTime - startTime));
    

    /**
     * 访问直接内存
     */
    public static void directAccess() 
        long startTime = System.currentTimeMillis();
        //分配直接内存
        ByteBuffer buffer = ByteBuffer.allocateDirect(1000);
        for (int i = 0; i < 100000; i++) 
            for (int j = 0; j < 200; j++) 
                buffer.putInt(j);
            
            buffer.flip();
            for (int j = 0; j < 200; j++) 
                buffer.getInt();
            
            buffer.clear();
        
        long endTime = System.currentTimeMillis();
        System.out.println("直接内存访问:" + (endTime - startTime));
    

    /**
     * 分配堆内存
     */
    public static void heapAllocate() 
        long startTime = System.currentTimeMillis();
        for (int i = 0; i < 100000; i++) 
            ByteBuffer.allocate(100);
        
        long endTime = System.currentTimeMillis();
        System.out.println("堆内存申请:" + (endTime - startTime));
    

    /**
     * 分配直接内存
     */
    public static void directAllocate() 
        long startTime = System.currentTimeMillis();
        for (int i = 0; i < 100000; i++) 
            ByteBuffer.allocateDirect(100);
        
        long endTime = System.currentTimeMillis();
        System.out.println("直接内存申请:" + (endTime - startTime));
    

    public static void main(String args[]) 
        for (int i = 0; i < 5; i++) 
            heapAccess();
            directAccess();
        
        System.out.println("============================");
        for (int i = 0; i < 5; i++) 
            heapAllocate();
            directAllocate();
        
    

通过运行结果可以看到:堆内存的访问效率低于直接内存,堆内存的分配速度高于直接内存

使用直接内存的优缺点:
优点:

  1. 不占用堆内存空间,减少了发生GC的可能
  2. java虚拟机实现上,本地IO会直接操作直接内存(直接内存=>系统调用=>硬盘/网卡),而非直接内存则需要二次拷贝(堆内存=>直接内存=>系统调用=>硬盘/网卡)

缺点:

  1. 初始分配较慢
  2. 没有JVM直接帮助管理内存,容易发生内存溢出。为了避免一直没有FULL GC,最终导致直接内存把物理内存被耗完。我们可以指定直接内存的最大值,通过-XX:MaxDirectMemorySize来指定,当达到阈值的时候,调用system.gc来进行一次FULL GC,间接把那些没有被使用的直接内存回收掉。

心跳检测

TCP在长连接时,会定期在客户端和服务器之间发送心跳包(一种很小的数据包),通知对方自己在线,一旦长期收不到心跳包,则认为对方离线,自己可以进行异常处理。

Netty的IdleStateHandler可以实现心跳检测机制

构造方法

IdleStateHandler(int readerIdleTime, int writerIdleTime, int allIdleTime,TimeUnit unit )

参数:

  1. readerIdleTime: 读超时. 即当在指定的时间间隔内没有从 Channel 读取到数据时会触发
  2. writerIdleTime: 写超时. 即当在指定的时间间隔内没有数据写入到 Channel 时会触发
  3. allIdleTime: 读/写超时. 即当在指定的时间间隔内没有读或写操作时会触发
  4. unit 时间单位

带心跳检测机制的服务器端

添加了IdleStateHandler,并设置读超时为3秒,3秒没有读取客户端消息则会触发下一个处理器中的userEventTriggered方法

public class HeartBeatServer 

    public static void main(String[] args) throws Exception 
        EventLoopGroup boss = new NioEventLoopGroup();
        EventLoopGroup worker = new NioEventLoopGroup();
        try 
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(boss, worker)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() 
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception 
                            ChannelPipeline pipeline = ch.pipeline();
                            pipeline.addLast("decoder", new StringDecoder());
                            pipeline.addLast("encoder", new StringEncoder());
                            //IdleStateHandler的readerIdleTime参数指定超过3秒还没收到客户端的连接,
                            //会触发IdleStateEvent事件并且交给下一个handler处理,下一个handler必须
                            //实现userEventTriggered方法处理对应事件
                            pipeline.addLast(new IdleStateHandler(3, 0, 0, TimeUnit.SECONDS));
                            pipeline.addLast(new HeartBeatServerHandler());
                        
                    );
            System.out.println("netty server start。。");
            ChannelFuture future = bootstrap.bind(9000).sync();
            future.channel().closeFuture().sync();
         catch (Exception e) 
            e.printStackTrace();
         finally 
            worker.shutdownGracefully();
            boss.shutdownGracefully();
        
    



服务器处理器

重写了userEventTriggered方法,执行心跳超时的处理逻辑

public class HeartBeatServerHandler extends SimpleChannelInboundHandler<String> 

    int readIdleTimes = 0;

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String s) throws Exception 
        System.out.println(" ====== > [server] message received : " + s);
        if ("Heartbeat Packet".equals(s)) 
            ctx.channel().writeAndFlush("ok");
         else 
            System.out.println(s);
        
    

    //触发超时事件
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception 
        IdleStateEvent event = (IdleStateEvent) evt;

        String eventType = null;
        switch (event.state()) 
            case READER_IDLE:
                eventType = "读空闲";
                readIdleTimes++; // 读空闲的计数加1
                break;
            case WRITER_IDLE:
                eventType = "写空闲";
                // 不处理
                break;
            case ALL_IDLE:
                eventType = "读写空闲";
                // 不处理
                break;
        
        System.out.println(ctx.channel().remoteAddress() + "超时事件:" + eventType);
        if (readIdleTimes > 3) 
            System.out.println(" [server]读空闲超过3次,关闭连接,释放更多资源");
            ctx.channel().writeAndFlush("idle close");
            ctx.channel().close();
        
    

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception 
        System.err.println("=== " + ctx.channel().remoteAddress() + " is active ===");
    

客户端定期发送心跳数据包

public class HeartBeatClient 
    public static void main(String[] args) throws Exception 
        EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
        try 
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(eventLoopGroup).channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() 
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception 
                            ChannelPipeline pipeline = ch.pipeline();
                            pipeline.addLast("decoder", new StringDecoder());
                            pipeline.addLast("encoder", new StringEncoder());
                            pipeline.addLast(new HeartBeatClientHandler());
                        
                    );

            System.out.println("netty client start。。");
            Channel channel = bootstrap.connect("127.0.0.1", 9000).sync().channel();
            String text = "Heartbeat Packet";
            Random random = new Random();
            //随机时间间隔发送心跳包
            while (channel.isActive()) 
                int num = random.nextInt(10);
                Thread.sleep(num * 1000);
                channel.writeAndFlush(text);
            
         catch (Exception e) 
            e.printStackTrace();
         finally 
            eventLoopGroup.shutdownGracefully();
        
    

    static class HeartBeatClientHandler extends SimpleChannelInboundHandler<String> 

        @Override
        protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception 
            System.out.println(" client received :" + msg);
            if (msg != null && msg.equals("idle close")) 
                System.out.println(" 服务端关闭连接,客户端也关闭");
                ctx.channel().closeFuture();
            
        
    



以上是关于Netty框架的高级概念的主要内容,如果未能解决你的问题,请参考以下文章

用Netty发生堆外内存泄露,看老司机一顿排查

Netty网络编程框架的核心概念以及入门案例

高性能网络通信框架Netty-基础概念篇

NIO高级编程与Netty入门

美团大牛熬夜亲码 Netty 全栈笔记太强了!入门 / 中级 / 高级 / 源码 / 架构全都有!

美团大牛熬夜亲码 Netty 全栈笔记太强了!入门 / 中级 / 高级 / 源码 / 架构全都有!