【测试开发】Mq消息重复如何测试？

本篇文章主要讲述重复消费的原因，以及如何去测试这个场景，最后也会告诉大家，目前互联网项目关于如何避免重复消费的解决方案。

Mq为什么会有重复消费的问题?

Mq 常见的缺点之一就是消息重复消费问题，产生这种问题的原因是什么呢？有以下几点：

工作流程

Mq消息重复如何测试？

1、producer 生成数据，发送到broker集群，当遇到网络抖动超时，可能会重复发送。

为了保证数据的可靠性一般都会配置重试机制如下：

rocketmq:
  producer:
    group: sanyouProducer
    #发送消息超过5秒未接收到broker返回的成功消息
    send-message-timeout: 5000
    #重试最大次数
    retry-times-when-send-failed: 2
    max-message-size: 4194304
  name-server: 172.30.34.10:9876;172.30.35.37:9876;172.30.35.30:9876
  #发送消息超时时长，意思是超过5秒钟未收到broker返回的发送成功的消息，
  #producer会重复发送，但并不是一直发送，会根据retry-times-when-send-failed次数，
  #最多重试多少次

极端情况下，网络出现抖动，生产者超过设置的时间未收到broker返回的成功消息，会重新发送消息。

2、消费者宕机，未提交offset给broker

由上图可知，broker接收到producer 发送的消息后，会把消息发送给消费者，一般情况下，消费者消费完一条数据，会提交一个offset给到broker，告诉它，这条消息我消费了，但是，极端情况下，消费者消费一条消息成功，提交offset之前，宕机了或者网络抖动超时了，broker未收到offset，就认为这条消息没人消费，当消费者重启服务器或网络恢复，那么broker还会发送这条消息给消费者重新消费。

3、业务上的bug，可能会导致重复消费。

生产者producer的上游系统，突然出现了bug,导致重复调用生产者所在服务的接口，生产者收到请求后，继续发送消息给broker。

当然了，重复消费的原因有很多，以上只是常见的几种原因，那怎么去测试呢？

怎么测试重复消费场景？

假如有这么一个场景，采购员在采购系统的前端页面进行采购单下单操作，下单成功后，采购系统这边会保留一份采购单数据，然后发送一条mq给到wms 仓库系统，那么生产者就是采购系统，消费者就是wms仓库系统，wms消费到采购单的消息，落入数据库wms_purchase表中，为了简化，我只设计了三个字段。

建表ddl：

CREATE TABLE `wms_purchase` (
  `id` bigint(11) NOT NULL AUTO_INCREMENT COMMENT '仓库采购单id',
  `purchase_id` bigint(20) NOT NULL COMMENT '采购单id',
  `purchase_name` varchar(255) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=237 DEFAULT CHARSET=utf8;

怎么测试呢？很简单，我们只要编写生产者工具，在工具里加个循环，尽量循环多次，如下：

@RestController
@RequestMapping("/mq")
public class ProducerController {
    // 自动注入 RocketMQTemplate模板类，用于生产消息
    @Autowired
    private RocketMQTemplate mqTemplate;
    // 模拟生产者重复消费问题，前提是数据库没有唯一索引，并且项目未做幂等性校验
    @RequestMapping("/send")
    public String testSend(@RequestBody WmsPurchaseDto params) {
        try {
            for (int i = 0; i <100 ; i++) {
                mqTemplate.convertAndSend("fourbrothertopic", params);
            }
            return "success";
        } catch (Exception e) {
            e.printStackTrace();
            return "fail";
        }
    }

解读：

requestmapping对外暴露一个web接口，地址是localhost:8080/demo/mq/send，
post请求，参数是json格式，类似
{
    "purchaseId": "256465",
    "purchaseName": "测试"
}
这种形式，然后起个for循环，循环调用convertAndSend方法，发送同样的消息，最终结果如下图：

Mq消息重复如何测试？

这里模拟producer重复发送的场景，前提是数据库没有对采购单id做唯一索引，并且项目未做幂等性校验。数据库里出现很多采购单id一样的数据，业务上这是不允许的。

假如说，项目出现了这么一种bug，开发那边是怎么修复的呢？

Mq如何保证幂等性？

分享几种解决方案的具体代码demo：

1、数据库unique key（表里不允许重复列出现）来保证幂等性。

很简单，我们只要在wms_purchase里，对purchaseId添加唯一索引即可，提示：在添加唯一索引之前，需清理完表里的数据。

也可以使用ddl语句：

ALTER TABLE `wms_purchase` ADD UNIQUE ( `purchaseId` )

代码不变，调用以下接口：

localhost:8080/demo/mq/send post请求
{
    "purchaseId": "256465",
    "purchaseName": "测试"
}

得到以下结果：

Mq消息重复如何测试？

上图中，循环生产同一条采购单数据，但是右边表中只出现了一条采购单id是256465的数据，说明添加唯一索引确实保证了幂等性，但是代码里却出现大量类似Duplicate entry '256465' for key 'uniqe_key_purchaseid' 日志，是因为触发了数据设置的唯一索引，

由于触发了唯一索引，导致消费者未提交offset给broker，那么broker会认为这条消息未被消费，后续会持续不断地推送消息给消费者，也就意味着会持续不断地报错。

另外这种持续无效的请求数据库会占用数据库的连接资源，在高并发的场景下，会严重拖垮系统响应效率。

虽然保证了幂等性，但是日志里总是报错，太不讲究、也不雅观，那怎么解决呢？

2、数据库unique key+redis 来保证幂等性。

如截图：

Mq消息重复如何测试？

通俗的理解就是，消费者在进行数据库落库操作之前，会判断redis是有这条采购单数据，如果有就直接放过这条消息不做处理，没有这条数据，那就进行落库操作，但在落库之前还要进一步判断数据库是否有这条采购单数据，没有那就进行落库，落库成功，再把采购单的id当做key,采购单数据当做value set 进redis缓存里，设置一定的过期时间。

redis基于内存，操作数据特别快，在进行落库之前查询redis，可以避免很多无效的请求数据库，但是为啥要设置过期时间？因为redis的内存资源有限，并且很宝贵，所以我们希望设置的数据能在一段时间内定期失效，即使失效，也没关系，还有数据库的唯一索引兜底。

这样就很好的保证了幂等性，也避免了大量的日志报错。伪代码如下：

@Component
//mq的监听器，指定topic是TopicTest，消费者组consumerGroupTest
@RocketMQMessageListener(topic = "fourbrothertopic", consumerGroup = "consumerGroupTest")
@Slf4j
public class ConsumeController implements RocketMQListener {

    @Autowired
    private WmsPurchaseMapper wmsPurchaseMapper;

    @Autowired
    private RedisTemplate redisTemplate;
    @Override
    public void onMessage(String message) {
        log.info("------- Consumer: {}", message);
        //将message消息映射成WmsPurchase实体
        WmsPurchase wmsPurchase = JSONObject.parseObject(message, WmsPurchase.class);
        //首先判断redis里面是否有这条采购单数据，通过PurchaseId查询，有数据，则直接放过不做处理
       if (redisTemplate.opsForValue().get(wmsPurchase.getPurchaseId().toString())==null){
           //然后再使用PurchaseId查询数据库，有数据，则直接放过不做处理
           if (null == wmsPurchaseMapper.selectByPurchaseId(wmsPurchase.getPurchaseId())){
               //数据库没有数据，就进行插入操作，
               if (wmsPurchaseMapper.insert(wmsPurchase)>0){
                   //插入成功就把purchaseid塞进redis里，过期时间是72小时
                   redisTemplate.opsForValue().set(wmsPurchase.getPurchaseId(),wmsPurchase.toString(),72, TimeUnit.HOURS);
               }
           }else {
               //能走到这个判断分支，说明缓存里的采购单数据已经失效，如果还有消息重复消费
               //那就再放入缓存一次，72h过期
               redisTemplate.opsForValue().set(wmsPurchase.getPurchaseId(),wmsPurchase.toString(),72, TimeUnit.HOURS);
               log.info("数据库已保留该数据");
               // 触发重复消费告警机制
           }
       }else {
           log.info("缓存已保留该数据");
            // 触发重复消费告警机制
       }
    }
}

思路很简单，如代码中注释。当然这种方法也有缺点，就是过于依赖redis，有些系统没有使用redis组件，那么还得维护一套redis组件，并且还得保证redis集群高可用。那项目只有mysql，能不能依靠数据库去维护保证幂等性呢？当然可以！

3、还有一种方法叫去重表+唯一索引，顾名思义就是另外维护一张表，记录已经消费的采购单数据，其实和上述方法差不多，上述方法查询缓存，取重表查询数据库取重表。

伪代码 如下：

@Component
//mq的监听器，指定topic是TopicTest，消费者组consumerGroupTest
@RocketMQMessageListener(topic = "fourbrothertopic", consumerGroup = "consumerGroupTest")
@Slf4j
public class ConsumeController implements RocketMQListener {

    @Autowired
    private WmsPurchaseMapper wmsPurchaseMapper;

    @Autowired
    private UniquePurchaseMapper uniquePurchaseMapper;

    @Autowired
    private RedisTemplate redisTemplate;
    @SneakyThrows
    @Override
    public void onMessage(String message) {
        log.info("------- Consumer: {}", message);
        //将message消息映射成WmsPurchase实体
        WmsPurchase wmsPurchase = JSONObject.parseObject(message, WmsPurchase.class);
        log.info("映射后实体消息"+ JSON.toJSONString(wmsPurchase));
        if (uniquePurchaseMapper.selectByPurchaseId(wmsPurchase.getPurchaseId().intValue())  == null){
            if (null == wmsPurchaseMapper.selectByPurchaseId(wmsPurchase.getPurchaseId())){
                //数据库没有数据，就进行插入操作，
                if (wmsPurchaseMapper.insert(wmsPurchase)>0){
                    //插入成功就把purchaseid塞进unique_purchase
                    UniquePurchase  uniquePurchase =   new UniquePurchase();
                    uniquePurchase.setPurchaseId(wmsPurchase.getPurchaseId().intValue());
                    log.info("插入取重表消息："+ JSON.toJSONString(uniquePurchase));
                    uniquePurchaseMapper.insert(uniquePurchase);
                }
            }else {
                log.info("数据库已保留该数据");
                //自动触发告警机制
            }
        }else {
            log.info("取重表已有这条采购单数据");
        }
 }

代码已上传至gitee,感兴趣可以自行阅读。

上述方式在查询取重表时，并发不安全，极端情况下还是会触发唯一索引错误，比如说，消费者要消费大量消息（线程），执行上述代码，A线程执行完23行，挂起了，cpu把执行权给了B线程，B执行到25行并插入成功，那么这时A线程被唤起，也执行到了23行，结果触发了唯一索引错误。那怎么避免呢？

我们可以让所有线程别并发执行，串行执行，那就用到redis的分布式锁技术。

4、分布式锁+uniquekey

伪代码如下

@Component
//mq的监听器，指定topic是TopicTest，消费者组consumerGroupTest
@RocketMQMessageListener(topic = "fourbrothertopic", consumerGroup = "consumerGroupTest")
@Slf4j
public class ConsumeController implements RocketMQListener {

    @Autowired
    private WmsPurchaseMapper wmsPurchaseMapper;
    @Autowired
    private RedissonClient redisson;
    @Autowired
    private UniquePurchaseMapper uniquePurchaseMapper;
    @Autowired
    private RedisTemplate redisTemplate;
    @SneakyThrows
    @Override
    public void onMessage(String message) {
        log.info("------- Consumer: {}", message);
        //将message消息映射成WmsPurchase实体
        WmsPurchase wmsPurchase = JSONObject.parseObject(message, WmsPurchase.class);
// 注入redisson
// 获取锁对象
        RLock lock = redisson.getLock("lockName");
        try {
            // 1. 最常见的使用方法
            //lock.lock();
            // 2. 支持过期解锁功能,10秒钟以后自动解锁, 无需调用unlock方法手动解锁
            //lock.lock(10, TimeUnit.SECONDS);
            // 3. 尝试加锁，最多等待2秒，上锁以后8秒自动解锁
            boolean res = lock.tryLock();
            if (res) { //成功
                    //然后再使用PurchaseId查询数据库，有数据，则直接放过不做处理
                    if (null == wmsPurchaseMapper.selectByPurchaseId(wmsPurchase.getPurchaseId())){
                        //数据库没有数据，就进行插入操作，
                        if (wmsPurchaseMapper.insert(wmsPurchase)>0){
                            //插入成功就把purchaseid塞进redis里，过期时间是72小时
                            redisTemplate.opsForValue().set(wmsPurchase.getPurchaseId().toString(),wmsPurchase.toString(),1, TimeUnit.HOURS);
                        }
                    }else {
                        redisTemplate.opsForValue().set(wmsPurchase.getPurchaseId().toString(),wmsPurchase.toString(),1, TimeUnit.HOURS);
                        log.info("数据库已保留该数据");
                        //自动触发告警机制
                    }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            //释放锁
            RLock lockName = redisson.getLock("lockName");
            if (lockName.isLocked()) {
                if (lockName.isHeldByCurrentThread()) {
                    lockName.unlock();
                }
            }
        }
}

这种也是比较常见的一种，缺点也很明显，在高并发，大请求量的场景下，所有线程串行执行，处理效率势必会降低。当然了，技术没有好坏，只有合不合适。如果你的项目并发量一般，可以尝试使用上述方法。

具体代码demo已上传至gitee平台，地址如下：

https://gitee.com/lv1792017548/rocketmq-demo.git

总结

本文主要分享了如何测试mq消息队列重复性消费，以及避免重复消费常见的解决方案。

【B站最全最易学】十年大佬终于将测试开发路线整理出来了，小白一学就会，拿走不谢，允许白嫖！！

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

如若内容造成侵权/违法违规/事实不符，请联系我们进行投诉反馈qq邮箱809451989@qq.com，一经查实，立即删除！

相关文章

16、外部配置源与外部配置文件及JSON配置

16、外部配置源与外部配置文件及JSON配置

外部配置源与外部配置文件及JSON配置 application.properties application.yml 这些是配置文件&#xff0c; 命令行配置、环境变量配置、系统属性配置源&#xff0c;这些属于配置源。 外部配置源的作用&#xff1a; Spring Boot相当于对Spring框架进行了封装&#xff0c;Spri…
阅读更多...
webrtc的回声消除延迟时间估算

webrtc的回声消除延迟时间估算

叫回声消除的延迟时间估算不太合理&#xff0c;这里核心就是估算调用webrtc的条件边界&#xff0c;都知道webrtc回声消除的生效的前提就是一定要拿到远端声音的信息&#xff0c;然后拿近端声音和远端声音对齐&#xff0c;从近端声音中&#xff0c;结合远端声音模拟出远端声音在…
阅读更多...
Windows用户如何安装新版本cpolar内网穿透超详细教程

Windows用户如何安装新版本cpolar内网穿透超详细教程

Windows用户如何安装新版本cpolar内网穿透 文章目录 Windows用户如何安装新版本cpolar内网穿透 在科学技术高度发达的今天&#xff0c;我们身边充斥着各种电子产品&#xff0c;这些电子产品不仅为我们的工作带来极大的便利&#xff0c;也让生活变得丰富多彩。我们可以使用便携的…
阅读更多...
[Python] Pylance 插件打开 Python 的类型检查

[Python] Pylance 插件打开 Python 的类型检查

安装 Python 插件 2.打开一个 Python 文件 可以看到右下角有一个花括号和 Python 字样&#xff0c;点击花括号&#xff08;不是 Python 字样&#xff09;打开类型检查即可&#xff1a;
阅读更多...
酷开系统 | 酷开科技，让数据变得更有价值！

酷开系统 | 酷开科技，让数据变得更有价值！

身处信息时代&#xff0c;我们每个人时刻都在生成、传递和应用数据&#xff0c;数据已经成为了现代社会中宝贵的资源之一&#xff0c;而在人工智能领域&#xff0c;数据更是被称为人工智能的“燃料”。 而在AI的发展中&#xff0c;只有拥有高质量、多样性且充分代表性的数据集…
阅读更多...
java 定时任务不按照规定时间执行

java 定时任务不按照规定时间执行

这里写目录标题 使用异步启动可能出现的问题排查代码中添加的定时任务步骤是否正确排查是否任务阻塞&#xff0c;如果定时任务出现异常阻塞后&#xff0c;将不会在次执行java中多个Scheduled定时器不执行为了让Scheduled效率更高&#xff0c;我们可以通过两种方法将定时任务变成…
阅读更多...
vxworks文件系统分析

vxworks文件系统分析

参考https://www.freebuf.com/articles/endpoint/335030.html 测试固件 https://service.tp-link.com.cn/detail_download_7989.html 固件提取 binwalk解压固件&#xff0c;在第一部分即为要分析的二进制文件&#xff0c;可以拖进ida分析 设置为arm小端字节序&#xff0c;点…
阅读更多...
【HarmonyOS】性能优化之低代码开发加载多张轮播图

【HarmonyOS】性能优化之低代码开发加载多张轮播图

【关键字】 HarmonyOS、低代码开发、Swiper组件、性能优化、分页加载 写在前面 目前使用DevEco Studio的低代码工具开发元服务时&#xff0c;通过实际测试发现&#xff0c;Swiper组件加载多张轮播图时加载显示耗时较长&#xff08;实际测试网络状态一般的情况下显示耗时达到8…
阅读更多...
ER系列路由器多网段划分设置指南

ER系列路由器多网段划分设置指南

ER系列路由器多网段划分设置指南 - TP-LINK 服务支持 TP-LINK ER系列路由器支持划分多网段&#xff0c;可以针对不同的LAN接口划分网段&#xff0c;即每一个或多个LAN接口对应一个网段&#xff1b;也可以通过一个LAN接口与支持划分802.1Q VLAN的交换机进行对接&#xff0c;实现…
阅读更多...
Stable Diffusion：网页版 体验 / AI 绘图

Stable Diffusion：网页版 体验 / AI 绘图

一、官网地址 Stable Diffusion Online 二、Stable Diffusion AI 能做什么 Stable Diffusion AI绘图是一种基于Stable Diffusion模型的生成式AI技术&#xff0c;能够生成各种类型的图像&#xff0c;包括数字艺术、照片增强和图像修复等。以下是一些可能的应用&#xff1a; …
阅读更多...
怎么设置文件夹密码？文件夹密码设置方法合集

怎么设置文件夹密码？文件夹密码设置方法合集

为文件夹设置密码可以有效地保护文件夹的数据安全&#xff0c;那么该怎么设置文件夹密码呢&#xff1f;下面我们来一起了解一下。 文件夹保护3000 想要简单快捷的为文件夹设置密码&#xff0c;那么&#xff0c;文件夹保护3000就是最佳的选择。它提供了3种文件夹保护方式&#…
阅读更多...
IDEA项目实践——创建Java项目以及创建Maven项目案例、使用数据库连接池创建项目简介

IDEA项目实践——创建Java项目以及创建Maven项目案例、使用数据库连接池创建项目简介

系列文章目录 IDEA上面书写wordcount的Scala文件具体操作 IDEA创建项目的操作步骤以及在虚拟机里面创建Scala的项目简单介绍 目录 系列文章目录 前言 一 准备工作 1.1 安装Maven 1.1.1 Maven安装配置步骤 1.1.2 解压相关的软件包 1.1.3 Maven 配置环境变量 1.1.4 配…
阅读更多...
vue2 el-carousel轮播图和文字一起改变

vue2 el-carousel轮播图和文字一起改变

vue项目的话 安装一下element依赖 npm i element-ui -S在main入口文件引入element包 我在app文件里边去写的 <template><div class"w"><el-carousel height"460px"><el-carousel-item v-for"item in items" :key"i…
阅读更多...
C++之观察者模式（发布-订阅）

C++之观察者模式（发布-订阅）

目录 模式简介 介绍 优点 缺点 代码实现 场景说明 实现代码 运行结果 模式简介 观察者模式&#xff08;Observer Pattern&#xff09;&#xff0c;也叫我们熟知的发布-订阅模式。 它是一种行为型模式。 介绍 观察者模式主要关注的是对象的一对多的关系&#xff0c; …
阅读更多...
C++ 哈希的应用【布隆过滤器】

C++ 哈希的应用【布隆过滤器】

✨个人主页&#xff1a; 北 海 &#x1f389;所属专栏&#xff1a; C修行之路 &#x1f383;操作环境&#xff1a; Visual Studio 2022 版本 17.6.5 文章目录 &#x1f307;前言&#x1f3d9;️正文1、字符串比较2、布隆过滤器的概念3、布隆过滤器的实现3.1、基本结构3.2、插入…
阅读更多...
实现邮箱管理之gmail邮箱、office365（Azure）邮箱之披荆斩棘问题一览

实现邮箱管理之gmail邮箱、office365（Azure）邮箱之披荆斩棘问题一览

要进行Office365邮箱的授权对接&#xff0c;你需要先申请一个应用&#xff0c;并获取授权访问令牌。 以下是一个简单的步骤&#xff1a; 登录 Azure 门户&#xff1a;https://portal.azure.com/创建一个新的应用程序&#xff0c;或者使用现有的应用程序。要创建新的应用程序&…
阅读更多...
SpringBoot引入MyBatisGenerator

SpringBoot引入MyBatisGenerator

1.引入插件 <plugin><groupId>org.mybatis.generator</groupId><artifactId>mybatis-generator-maven-plugin</artifactId><version>1.3.5</version><configuration><!--generator配置文件所在位置--><configuratio…
阅读更多...
【CSS】3D卡片效果

【CSS】3D卡片效果

效果 index.html <!DOCTYPE html> <html><head><title> Document </title><link type"text/css" rel"styleSheet" href"index.css" /></head><body><div class"card"><img…
阅读更多...
Spring：IOC技术、Bean、DI

Spring：IOC技术、Bean、DI

前言 Spring是一个开源的项目&#xff0c;并不是单单的一个技术&#xff0c;发展至今已形成一种开发生态圈。也就是说我们可以完全使用Spring技术完成整个项目的构建、设计与开发。Spring是一个基于IOC和AOP的架构多层j2ee系统的架构。 SpringFramework&#xff1a;Spring框架…
阅读更多...
Jenkins Gerrit Trigger实践

Jenkins Gerrit Trigger实践

1.创建Gerrit Trigger 2.jenkins master节点生成gerrit用户的密钥 这里的用户名得写登录gerrit后个人信息中的 Username 3.gerrit 配置刚刚jenkins生成密钥的公钥 4.gerrit 用户加入群组 不加这个群组&#xff0c;下一步测试就会报错“User aeshare has no capability conn…
阅读更多...
最新文章

Copyright @ 2022~2023