Netty之使用DelimiterBasedFrameDecoder进行消息分隔详解

更新时间：2023年12月14日 09:59:00 作者：Terisadeng

这篇文章主要介绍了Netty之使用DelimiterBasedFrameDecoder进行消息分隔详解,在使用Netty进行TCP消息传输时,为了上层协议能够对消息正确区分,避免粘包和拆包导致的问题,一般可以通过消息定长、将回车换行符作为消息结束符,需要的朋友可以参考下

DelimiterBasedFrameDecoder消息分隔

在使用Netty进行TCP消息传输时，为了上层协议能够对消息正确区分，避免粘包和拆包导致的问题。

一般可以通过消息定长、将回车换行符作为消息结束符、将特殊的分隔符作为消息的结束标志或者在消息头中定义长度字段来标识消息的总长度。

其中常用的通过分隔符作为消息的结束标志就涉及到Netty的DelimiterBasedFrameDecoder类,服务端如下：

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
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;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
public class EchoServer
{
    public void bind(int port)throws Exception{
        //配置服务端的NIO线程组
        EventLoopGroup bossGroup=new NioEventLoopGroup();
        EventLoopGroup workerGroup=new NioEventLoopGroup();
        try
        {
            ServerBootstrap b=new ServerBootstrap();
            b.group(bossGroup, workerGroup)
            .channel(NioServerSocketChannel.class)
            .option(ChannelOption.SO_BACKLOG, 1024)
            //控制台输出服务端运行日志
            .handler(new LoggingHandler(LogLevel.INFO))
            //编写服务端接收和发送消息的具体逻辑
            .childHandler(new ChildChannleHandler());
            //绑定启动端口，同步等待成功
            ChannelFuture f=b.bind(port).sync();
            //等待服务端监听端口关闭
            f.channel().closeFuture().sync();
        }
        catch (Exception e)
        {
            e.printStackTrace();
        }
        finally{
            //释放线程资源
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
    //服务端接收到客户端的消息时会先执行该类的initChannel()方法进行channel的初始化操作
    private class ChildChannleHandler extends ChannelInitializer<SocketChannel>{
        @Override
        protected void initChannel(SocketChannel arg0)
            throws Exception
        {
            //创建分隔符缓冲对象，使用"$_"作为分隔符
            ByteBuf delimiter=Unpooled.copiedBuffer("$_".getBytes());
            //创建DelimiterBasedFrameDecoder对象，将其加入到ChannelPipeline
            //参数1024表示单条消息的最大长度，当达到该长度仍然没有找到分隔符就抛出TooLongFrame异常，第二个参数就是分隔符
            //由于DelimiterBasedFrameDecoder自动对请求消息进行了解码，下面的ChannelHandler接收到的msg对象就是完整的消息包
            arg0.pipeline().addLast(new DelimiterBasedFrameDecoder(1024, delimiter));
            //StringDecoder解码器将ByteBuf解码成字符串对象，这样在ChannelHandlerAdapter中读取消息时就不需要通过ByteBuf获取了
            arg0.pipeline().addLast(new StringDecoder());
            //对网络事件进行读写操作的类
            arg0.pipeline().addLast(new EchoServerHandler());
        }
    }
    public static void main(String[] args)throws Exception
    {
        int port =8888;
        if (args!=null&&args.length>0)
        {
            port=Integer.valueOf(args[0]);
        }
        new EchoServer().bind(port);
    }
}

服务端消息读写操作：

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
//网络I/O事件读写操作
public class EchoServerHandler extends ChannelHandlerAdapter
{
    int counter=0;
    //接收客户端发送的消息并返回响应
    @Override
    public void channelRead(ChannelHandlerContext ctx,Object msg)throws Exception{
        //获取String类型的请求消息（StringDecoder已经对消息进行解码）
        String body=(String)msg;
        System.out.println("This is "+ ++counter+"times receive client : ["+body+"]");
        //由于设置了DelimiterBasedFrameDecoder过滤掉了分隔符"$_"，   因此需要将返回消息尾部拼接上分隔符
        body+="$_";
        //将接收到的消息再放到ByteBuf中重新发送给客户端
        ByteBuf buf=Unpooled.copiedBuffer(body.getBytes());
        //把待发送的消息放到发送缓冲数组中，并把缓冲区中的消息全部写入SockChannel发送给客户端
        ctx.writeAndFlush(buf);
    }
    //发生异常时关闭ChannelHandlerContext，释放和ChannelHandlerContext相关联的句柄等资源
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx,Throwable cause){
        cause.printStackTrace();
        ctx.close();
    }
}

客户端：

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
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.NioSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;
//客户端
public class EchoClient
{
    public void connect(int port,String host)throws Exception{
        //创建客户端进行I/O读写的线程组
        EventLoopGroup g=new NioEventLoopGroup();
        try
        {
            //创建客户端启动辅助类Bootstrap
            Bootstrap b=new Bootstrap();
            b.group(g)
            //设置Channel
            .channel(NioSocketChannel.class)
            //配置Channel
            .option(ChannelOption.TCP_NODELAY, true)
            //添加处理类，这里为了方便直接使用了匿名内部类
            .handler(new ChannelInitializer<SocketChannel>()
            {
                //当创建NioSocketChannel成功后，将ChannelHandler设置到ChannelPipeline中处理网络I/O事件
                @Override
                protected void initChannel(SocketChannel arg0)
                    throws Exception
                {
                    //与服务端相同，需要配置一系列的ChannelHandler
                    ByteBuf delimiter=Unpooled.copiedBuffer("$_".getBytes());
                    arg0.pipeline().addLast(new DelimiterBasedFrameDecoder(1024,delimiter));
                    arg0.pipeline().addLast(new StringDecoder());
                    //客户端的处理类加入ChannelPipeline
                    arg0.pipeline().addLast(new EchoClientHandler());
                }
            });
            //调用connect方法发起异步连接，并调用同步方法等待连接成功
            ChannelFuture f=b.connect(host, port).sync();
            //f.channel().writeAndFlush(Unpooled.wrappedBuffer("111$_".getBytes()));
            //等待客户端连接关闭
            f.channel().closeFuture().sync();
        }
        catch (Exception e)
        {
            e.printStackTrace();
        }
        finally{
            //释放线程组
            g.shutdownGracefully();
        }
    }
    public static void main(String[] args)throws Exception
    {
        int port=8888;
        if (args!=null&&args.length>0)
        {
            port=Integer.valueOf(args[0]);
        }
        new EchoClient().connect(port, "127.0.0.1");
    }
}

客户端网络I/O事件处理：

import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerAdapter;
import io.netty.channel.ChannelHandlerContext;
//客户端读写网络I/O事件类
public class EchoClientHandler extends ChannelHandlerAdapter
{
    int counter;
    //发送到服务端的消息，注意结尾的分隔符一定要和服务端配置的分隔符一致，否则服务端ChannelInitializer.initChannel()方法虽然能够调用，但是DelimiterBasedFrameDecoder无法找到分隔符，不会调用读取消息的channelRead方法
    static final String ECHO_REQ="Hi,Welcome to Netty.$_";
    public EchoClientHandler(){
    }
    //客户端发送消息的方法
    @Override
    public void channelActive(ChannelHandlerContext ctx)throws Exception{
        for (int i = 0; i < 10; i++ )
        {
            //Unpooled.copiedBuffer()方法是深克隆，也可以使用Unpooled.buffer()写入消息发送
            ctx.writeAndFlush(Unpooled.copiedBuffer(ECHO_REQ.getBytes()));
        }
    }
    //读取服务端发送的消息
    @Override
    public void channelRead(ChannelHandlerContext ctx,Object msg)throws Exception{
        String body=(String)msg;
        System.out.println("This is "+ ++counter+" times receive server:["+body+"]");
    }
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx)throws Exception{
        //将消息发送队列中的消息写入到SocketChannel中发送给对方，channelActive使用了writeAndFlush这里可以不重写
        ctx.flush();
    }
    //异常处理，关闭ChannelHandlerContext
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx,Throwable cause){
        cause.printStackTrace();
        ctx.close();
    }
}

启动服务端：

启动客户端发送消息：

到此这篇关于Netty之使用DelimiterBasedFrameDecoder进行消息分隔详解的文章就介绍到这了,更多相关DelimiterBasedFrameDecoder进行消息分隔内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:

MyBatis框架底层的执行原理源码解析
这篇文章主要介绍了MyBatis框架底层的执行原理源码解析,本文通过图文实例代码相结合给大家介绍的非常详细，对大家的学习或工作具有一定的参考借鉴价值，需要的朋友可以参考下
2022-06-06
几种常见的Java运行时异常小结
在Java编程语言中异常处理是一项关键的机制,它帮助开发者识别和修复程序运行时可能出现的问题,下面这篇文章主要给大家介绍了几种常见的Java运行时异常的相关资料,文中通过代码介绍的非常详细,需要的朋友可以参考下
2024-08-08
Java AQS 原理与 ReentrantLock 实现方法
AQS 的作用是解决同步器的实现问题,它将复杂的同步器实现分解为简单的框架方法,开发者只需要实现少量特定的方法就能快速构建出可靠的同步器,这篇文章主要介绍Java AQS原理与ReentrantLock实现,需要的朋友可以参考下
2025-03-03
通过实例了解java spring使用构造器注入的原因
这篇文章主要介绍了通过实例了解spring使用构造器注入的原因,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
2019-12-12
Java运算符从见过到掌握上
计算机的最基本用途之一就是执行数学运算，作为一门计算机语言，Java也提供了一套丰富的运算符来操纵变量，本篇对大家的学习或工作具有一定的价值，需要的朋友可以参考下
2021-09-09
详解如何使用Spring的@FeignClient注解实现通信功能
SpringBoot是一个非常流行的Java框架,它提供了一系列工具来使这种交互无缝且高效,在这些工具中,@FeignClient注解因其易用性和强大的功能而脱颖而出, 在这篇文章中,我们将探讨如何使用Spring的@FeignClient注解进行客户端-服务器通信,需要的朋友可以参考下
2023-11-11
解决tomcat启动时报Junit相关错误java.lang.ClassNotFoundException:
这篇文章主要介绍了解决tomcat启动时报Junit相关错误java.lang.ClassNotFoundException: org.junit.Test问题,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教
2025-05-05
Java邮件发送超时时间过长问题的优化方案
邮件发送是许多Java应用中常见的功能,尤其在用户注册、密码重置和系统通知中,对于一个高并发的系统来说,邮件发送的超时问题可能导致应用性能的下降,甚至影响用户体验,因此,本期我们将讨论Java邮件发送超时时间过长的问题,并深入探讨其成因和优化策略
2025-06-06
用java将GBK工程转为uft8的方法实例
本篇文章主要介绍了用java将GBK工程转为uft8的方法实例，小编觉得挺不错的，现在分享给大家，也给大家做个参考。一起跟随小编过来看看吧
2017-08-08
SpringBoot+Redis队列实现Java版秒杀的示例代码
本文主要介绍了SpringBoot+Redis队列实现Java版秒杀的示例代码，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
2023-06-06

Netty之使用DelimiterBasedFrameDecoder进行消息分隔详解

DelimiterBasedFrameDecoder消息分隔

相关文章

最新评论

大家感兴趣的内容

最近更新的内容

常用在线小工具