SpringBoot 2.7 集成 Netty 4 解决粘包半包问题

news2024/11/25 6:33:01

文章目录

    • 1 摘要
    • 2 核心代码
      • 2.1 Netty 服务端连接器
      • 2.2 Netty 客户端连接器
      • 2.3 Netty 服务端 Handler
      • 2.4 Netty 客户端 Handler
    • 3 推荐参考资料
    • 4 Github 源码

1 摘要

Netty 的粘包和半包问题是由于 Netty 在接收消息时无法判断消息是否发送完毕，只能靠读取消息时是否读满缓存来终止，因此就出现了连续发送多条消息，实际上 Netty 接收端只在一次读取这就是粘包；又或者一次发送的消息太长，读取的时候会丢弃一部分，这就是半包。为了解决这个问题，保证一次发送对应一次读取，类似于 http 请求，一次请求对应一次回复。这里介绍使用LengthFieldBasedFrameDecoder解决 Netty 粘包半包问题。

关于 LengthFieldBasedFrameDecoder 解码器类，这是用来确定接收端如何读取消息的，数据会请求头+请求体，请求头部分用来存放固定字段以及请求体的长度，在读取消息的时候拿到字段的长度来读取对应的消息体，这样就能完整地区分一条消息。

解码器需要与编码器搭配，LengthFieldPrepender 即为编码器，指在发送消息的时候添加固定长度消息头。

关于 Netty 入门教程，可参考:

SpringBoot 2.7 集成 Netty 4 模拟服务端与客户端通讯入门教程

2 核心代码

2.1 Netty 服务端连接器

demo-netty-server/src/main/java/com/ljq/demo/springboot/netty/server/init/InitStrongNettyServer.java

package com.ljq.demo.springboot.netty.server.init;

import com.ljq.demo.springboot.netty.server.handler.DemoStrongNettyServerHandler;
import io.netty.bootstrap.ServerBootstrap;
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.NioServerSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.net.InetSocketAddress;

/**
 * @Description: 强化版 netty 服务端
 * @Author: junqiang.lu
 * @Date: 2023/8/23
 */
@Slf4j
@Component
public class InitStrongNettyServer implements ApplicationRunner {

    @Value("${netty.port2:9125}")
    private Integer nettyPort;


    @Resource
    private DemoStrongNettyServerHandler nettyServerHandler;

    @Override
    public void run(ApplicationArguments args) throws Exception {
        this.start();
    }

    /**
     * 启动服务
     *
     * @throws InterruptedException
     */
    public void start() throws InterruptedException {
        // 连接管理线程池
        EventLoopGroup mainGroup = new NioEventLoopGroup(2);
        // 工作线程池
        EventLoopGroup workGroup = new NioEventLoopGroup(8);
        ServerBootstrap bootstrap = new ServerBootstrap();
        bootstrap.group(mainGroup, workGroup)
                // 指定 nio 通道
                .channel(NioServerSocketChannel.class)
                // 指定 socket 地址和端口
                .localAddress(new InetSocketAddress(nettyPort))
                // 添加子通道 handler
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        socketChannel.pipeline()
                                // 消息解码: 读取消息头和消息体
                                .addLast(new LengthFieldBasedFrameDecoder(4096, 0, 4, 0, 4))
                                // 消息编码: 将消息封装为消息头和消息体,在响应字节数据前面添加消息体长度
                                .addLast(new LengthFieldPrepender(4))
                                // 字符串编解码器
                                .addLast(new StringEncoder())
                                .addLast(new StringDecoder())
                                .addLast(nettyServerHandler);
                    }
                });
        // 异步绑定服务器，调用sync()方法阻塞等待直到绑定完成
        bootstrap.bind().sync();
        log.info("---------- [init] strong netty server start ----------");
    }
}

代码说明: new LengthFieldBasedFrameDecoder(4096, 0, 4, 0, 4) 指的是一次读取消息的最大长度是 4096 字节，消息长度字段为 4 个字节。

new LengthFieldPrepender(4) 指的是再发送消息时将前边 4 个字节来表示消息的长度。

注意事项:

                // 添加子通道 handler
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        socketChannel.pipeline()
                                // 消息解码: 读取消息头和消息体
                                .addLast(new LengthFieldBasedFrameDecoder(4096, 0, 4, 0, 4))
                                // 消息编码: 将消息封装为消息头和消息体,在响应字节数据前面添加消息体长度
                                .addLast(new LengthFieldPrepender(4))
                                // 字符串编解码器
                                .addLast(new StringEncoder())
                                .addLast(new StringDecoder())
                                .addLast(nettyServerHandler);

以上编解码的顺序一定不能错，定义消息读取的编解码器一定是放在最开始，如果不这样，依旧会出现粘包半包问题。

2.2 Netty 客户端连接器

demo-netty-server/src/main/java/com/ljq/demo/springboot/netty/server/client/DemoStrongNettyClient.java

package com.ljq.demo.springboot.netty.server.client;

import com.ljq.demo.springboot.netty.server.handler.DemoStrongNettyClientHandler;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
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;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;

/**
 * @Description: 强化版 Netty 客户端模拟器
 * @Author: junqiang.lu
 * @Date: 2023/8/23
 */
@Slf4j
public class DemoStrongNettyClient {

    private final String host;

    private final int port;

    private final EventLoopGroup mainGroup;

    private final Bootstrap bootstrap;

    private Channel channel;

    public DemoStrongNettyClient(String host, int port) {
        this.host = host;
        this.port = port;
        this.mainGroup = new NioEventLoopGroup();
        this.bootstrap = new Bootstrap();

    }

    public Channel getChannel() {
        return this.channel;
    }

    /**
     * 创建连接
     */
    public void connect() throws InterruptedException {
        bootstrap.group(mainGroup)
                .channel(NioSocketChannel.class)
                .remoteAddress(new InetSocketAddress(host, port))
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        socketChannel.pipeline()
                                // 消息解码: 读取消息头和消息体
                                .addLast(new LengthFieldBasedFrameDecoder(4096,0,4,0,4))
                                // 消息编码: 将消息封装为消息头和消息体,在响应字节数据前面添加消息体长度
                                .addLast(new LengthFieldPrepender(4))
                                // 字符串编解码器
                                .addLast(new StringEncoder())
                                .addLast(new StringDecoder())
                                .addLast(new DemoStrongNettyClientHandler());
                    }
                });
        ChannelFuture future = bootstrap.connect().sync();
        this.channel = future.channel();
    }

    /**
     * 发送消息
     *
     * @param message
     */
    public void sendMessage(String message) {
        log.info("客户端待发送消息:{}", message);
        Channel channel = this.getChannel();
        channel.writeAndFlush(message);
    }

    public void close() throws InterruptedException {
        log.info("关闭客户端");
        mainGroup.shutdownGracefully();
    }


    public static void main(String[] args) throws InterruptedException {
        String serverHost = "127.0.0.1";
        int serverPort = 9125;
        String message = "abcde啊哈哈哈";
        DemoStrongNettyClient nettyClient = new DemoStrongNettyClient(serverHost, serverPort);
        nettyClient.connect();
        for (int i = 0; i < 100000; i++) {
            nettyClient.sendMessage(message + i);
        }
        log.info("--------开始休眠 5 秒------------");
        Thread.sleep(5000L);
        log.info("--------休眠 5 秒结束------------");
        for (int i = 0; i < 5; i++) {
            nettyClient.sendMessage(i + message);
            Thread.sleep(100L);
        }
        nettyClient.close();
    }

}

代码说明：编解码器一定是客户端与服务端搭配使用的。

这里也添加了测试方法，实测单机 10W 并发没有问题。

2.3 Netty 服务端 Handler

Handler 这里也贴出来，主要是推荐使用线程池来异步处理事务，这样可以提高并发性能。

demo-netty-server/src/main/java/com/ljq/demo/springboot/netty/server/handler/DemoStrongNettyServerHandler.java

package com.ljq.demo.springboot.netty.server.handler;

import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.util.concurrent.DefaultThreadFactory;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @Description: 强化版 netty 服务端事务处理器
 * @Author: junqiang.lu
 * @Date: 2023/8/23
 */
@Slf4j
@Component
// 标记该类实例可以被多个 channel 共享
@ChannelHandler.Sharable
public class DemoStrongNettyServerHandler extends SimpleChannelInboundHandler<String> {

    /**
     * 工作线程池
     */
    private final ExecutorService executorService = new ThreadPoolExecutor(4, 8, 60, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(10000), new DefaultThreadFactory("strong-netty-work-pool"),
            new ThreadPoolExecutor.CallerRunsPolicy());


    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, String s) throws Exception {
        // 打印接收到的消息(观察 netty work 线程)
        log.info("strong netty server received message: {}", s);
        executorService.execute(() -> {
            // 异步处理业务(观察线程池线程)
            log.info("consume message async: {}", s);
            try {
                Thread.sleep(5);
            } catch (InterruptedException e) {
                log.error("work thread sleep error.", e);
            }
        });
    }
}

2.4 Netty 客户端 Handler

客户端的 Handler 就不需要添加线程池了，客户端主要是发送多，接收少

demo-netty-server/src/main/java/com/ljq/demo/springboot/netty/server/handler/DemoStrongNettyClientHandler.java

package com.ljq.demo.springboot.netty.server.handler;

import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

/**
 * @Description: 强化版 netty 客户端事务处理器
 * @Author: junqiang.lu
 * @Date: 2023/8/23
 */
@Slf4j
@Component
// 标记该类实例可以被多个 channel 共享
@ChannelHandler.Sharable
public class DemoStrongNettyClientHandler extends SimpleChannelInboundHandler<String> {


    /**
     * 接收消息
     *
     * @param channelHandlerContext
     * @param s
     * @throws Exception
     */
    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, String s) throws Exception {
        log.info("strong netty client receive server message: {}", s);
    }

    /**
     * 通道激活，创建连接
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        log.info("strong netty client connect to server.");
    }
}

3 推荐参考资料

Netty如何解决粘包半包问题

4 Github 源码

Gtihub 源码地址 : https://github.com/Flying9001/springBootDemo/tree/master/demo-netty-server

个人公众号:404Code,分享半个互联网人的技术与思考，感兴趣的可以关注.
404Code

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/922929.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

每天一分享#读up有感#

每天一分享#读up有感#

不知道开头怎么写&#xff0c;想了一下&#xff0c;要不&#xff0c;就这样吧&#xff0c;开头也就写完 今日分享 分享一博主的分享——https://blog.csdn.net/zhangay1998/article/details/121736687 全程高能&#xff0c;大佬就diao&#xff0c;一鸣惊人、才能卓越、名扬四…
阅读更多...
算法通关村十二关 | 字符串前缀问题

算法通关村十二关 | 字符串前缀问题

1. 最长公共前缀 题目&#xff1a;LeetCode14&#xff0c;14. 最长公共前缀 - 力扣&#xff08;LeetCode&#xff09; 思路一 我们先看公共前缀有什么特点。 第一种方式&#xff0c;竖着比较&#xff0c;如图左边所示&#xff0c;选取数组中第一个字符串的位置&#xff0c;每…
阅读更多...
platform相关资料

platform相关资料

Step 1: Hardware Settings for Vitis Platform — Vitis™ Tutorials 2021.2 documentationhttps://xilinx.github.io/Vitis-Tutorials/2021-2/build/html/docs/Vitis_Platform_Creation/Introduction/03_Edge_VCK190/step1.html https://www.cnblogs.com/VagueCheung/p/1313…
阅读更多...
Tensoeboard的一些坑与技巧

Tensoeboard的一些坑与技巧

安装 pip install tensorboard 安装过程中遇到tensoeboard.exe找不到&#xff0c;参考解决&#xff1a; https://blog.csdn.net/weixin_44532467/article/details/123525891 3.启动tensorboard 默认路径 &#xff08;&#xff09; tensorboard --logdir logstensorboard -…
阅读更多...
淘宝API技术解析，实现获得淘宝APP商品详情原数据

淘宝API技术解析，实现获得淘宝APP商品详情原数据

淘宝API&#xff08;Application Programming Interface&#xff09;是为开发者提供的一组接口&#xff0c;用于与淘宝平台进行数据交互。通过使用淘宝API&#xff0c;开发者可以获得淘宝平台上商品、店铺、订单等各种数据&#xff0c;并进行相应的业务操作。 要实现获取淘宝A…
阅读更多...
利用Kettle进行SQLServer与Oracle之间的数据迁移实践

利用Kettle进行SQLServer与Oracle之间的数据迁移实践

待更新 https://it.cha138.com/tech/show-1275283.html
阅读更多...
基于spring boot校园疫情信息管理系统/疫情管理系统

基于spring boot校园疫情信息管理系统/疫情管理系统

摘要 随着计算机技术&#xff0c;网络技术的迅猛发展&#xff0c;Internet 的不断普及&#xff0c;网络在各个领域里发挥了越来越重要的作用。特别是随着近年人民生活水平不断提高&#xff0c;校园疫情信息管理系统给学校带来了更大的帮助。 由于当前疫情防控形势复杂&#xff…
阅读更多...
史上最全软件测试入门到精通【测试+测开】

史上最全软件测试入门到精通【测试+测开】

测试学习大纲梳理 根据本人过往学习经验与理解&#xff0c;整理了一些关于测试学习内容与顺序&#xff0c;涵盖了基本软件测试工程师需要掌握的所有技能&#xff0c;希望可以给想了解的小伙伴们一些指引与帮助&#xff0c;有错误或需求的欢迎留言指出~ 一、web开发者模式 这…
阅读更多...
[LeetCode周赛复盘] 第 359 场周赛20230820

[LeetCode周赛复盘] 第 359 场周赛20230820

[LeetCode周赛复盘] 第 359 场周赛20230820 一、本周周赛总结2828. 判别首字母缩略词1. 题目描述2. 思路分析3. 代码实现 2829. k-avoiding 数组的最小总和1. 题目描述2. 思路分析3. 代码实现 2830. 销售利润最大化1. 题目描述2. 思路分析3. 代码实现 2831. 找出最长等值子数组…
阅读更多...
【智算中心】国产GPU横向对比

【智算中心】国产GPU横向对比

近日&#xff0c;沐曦发布了一篇名为《沐曦与智谱AI完成兼容性测试 共建软硬件一体化解决方案》的公众号&#xff0c;表示曦云C500千亿参数AI大模型训练及通用计算GPU与智谱AI开源的中英双语对话语言模型ChatGLM2-6B完成适配。测试结果显示&#xff0c;曦云C500在智谱AI的升级版…
阅读更多...
V4L2+单色USB摄像头编程实践3：读取MJPG格式图像

V4L2+单色USB摄像头编程实践3：读取MJPG格式图像

查看摄像头支持的MJPG格式的分辨率和帧率&#xff1a; $ v4l2-ctl --list-formats-ext --device /dev/video0 ioctl: VIDIOC_ENUM_FMTType: Video Capture[0]: MJPG (Motion-JPEG, compressed)Size: Discrete 1280x720 Interval: Discrete 0.008s (120.000 fps)Interval:…
阅读更多...
敏捷研发管理软件及敏捷管理流程

敏捷研发管理软件及敏捷管理流程

Scrum中非常强调公开、透明、直接有效的沟通&#xff0c;这也是“可视化的管理工具”在敏捷开发中如此重要的原因之一。通过“可视化的管理工具”让所有人直观的看到需求&#xff0c;故事&#xff0c;任务之间的流转状态&#xff0c;可以使团队成员更加快速适应敏捷开发流程。 …
阅读更多...
Java cc链2 分析

Java cc链2 分析

环境 cc4 <!-- https://mvnrepository.com/artifact/org.apache.commons/commons-collections4 --> <dependency><groupId>org.apache.commons</groupId><artifactId>commons-collections4</artifactId><version>4.0</version&…
阅读更多...
损失函数介绍

损失函数介绍

用softmax&#xff0c;就可以将一个输出值转换到概率取值的一个范围。 交叉熵损失CrossEntropyLoss 第一个参数weight&#xff0c; 各类别的loss设置权值&#xff0c; 如果类别不均衡的时候这个参数很有必要了&#xff0c;加了之后损失函数变成这样&#xff1a; 第二个参数ign…
阅读更多...
8年测试经验之谈 —— 什么是全链路压测？

8年测试经验之谈 —— 什么是全链路压测？

随着互联网技术的发展和普及&#xff0c;越来越多的互联网公司开始重视性能压测&#xff0c;并将其纳入软件开发和测试的流程中。 阿里巴巴在2014 年双11 大促活动保障背景下提出了全链路压测技术&#xff0c;能更好的保障系统可用性和稳定性。 什么是全链路压测&#xff1f; …
阅读更多...
湘潭大学 湘大 XTU 1251 Colombian Number 题解（非常详细）

湘潭大学 湘大 XTU 1251 Colombian Number 题解（非常详细）

参考文章 1.XTUOJ-1251-Colombian Number 链接 1251 题面 题目描述 对于正整数n,不存在整数k,使得n等于k加上k的数码累加和&#xff0c;我们称这样的数是哥伦比亚数或者自我数。 比如 11就不是一个哥伦比亚数&#xff0c;因为10加上10的数码累加和1等于11;而20则是一个哥伦…
阅读更多...
uniapp 开发微信小程序使用echart的dataZoom属性缩放功能不生效！bug记录！

uniapp 开发微信小程序使用echart的dataZoom属性缩放功能不生效！bug记录！

在本项目中使用的是这个echart库 在项目中添加了dataZoom配置项但是不生效&#xff0c;突然想到微信小程序代码大小的限制&#xff0c;之前的echarts.js是定制的&#xff0c;有可能没有加dataZoom组件。故重新定制echarts.js。之前用的echarts版本是5.0.0&#xff0c;这次也是…
阅读更多...
容灾设备系统组成，容灾备份系统组成包括哪些

容灾设备系统组成，容灾备份系统组成包括哪些

随着信息技术的快速发展&#xff0c;企业对数据的需求越来越大&#xff0c;数据已经成为企业的核心财产。但是&#xff0c;数据安全性和完整性面临巨大挑战。在这种环境下&#xff0c;容灾备份系统应运而生&#xff0c;成为保证企业数据安全的关键因素。下面我们就详细介绍容灾…
阅读更多...
IDEA启动Tomcat两个端口的方式 使用nginx进行反向代理 JMeter测试分布式情况下synchronized锁失效

IDEA启动Tomcat两个端口的方式 使用nginx进行反向代理 JMeter测试分布式情况下synchronized锁失效

目录 引出IDEA启动Tomcat两个端口的方式1.编辑配置2.添加新的端口-Dserver.port80833.service里面管理4.启动后进行测试 使用nginx进行反向代理反向代理多个端口运行日志查看启动关闭重启 分布式情况下synchronized失效synchronized锁代码启动tomcat两个端口nginx反向代理JMete…
阅读更多...
慕课网 Go工程师 第三周 package和gomodules章节

慕课网 Go工程师 第三周 package和gomodules章节

Go包的引入&#xff1a; 包名前面加匿名&#xff0c;只引入但不使用&#xff0c;如果对应包有init函数&#xff0c;会执行init函数&#xff08;初始化操作&#xff09; 包名前面加. 把这个包的结构体和方法导入当前包&#xff0c;慎用&#xff0c;你不知道当前包和被引入的包用…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023