Spring Boot项目实战：消息丢失和重复消费问题

发表于

你好，我是田哥

在我的充电桩项目中，有个用户积分模块，原型图如下：

图片

我的积分

下面来聊聊这个项目中，积分增加场景。

  • 用户充值完成后，赠送积分，比如：充100元，给用户积分新增100个。
  • 用户充电支付完成（非余额支付），赠送积分。
  • 用户邀请新人注册，赠送积分
  • 新用户认证完成，赠送积分
  • ….

关于积分增加策略，基本上都是由运营来定，总之，很多项目中都有这么个功能。

常规系统就是用户上面的行为伴随着用户积分处理完成（在同一个事物里），但，咱们为了提升系统性能和用户体验，我们通常把积分增加这类业务采用异步方式去实现。

比如：线程池、各种消息队列等

在我们这个版本中，采用的是RabbitMQ消息队列来实现的。

问题

既然使用RabbitMQ，那我们就不得不考虑关于消息的问题：

  • 消息丢失问题
  • 重复消费问题

消息丢失问题

这里对用户积分增加，如果把消息搞丢了，用户积分最终并不会得到增加，那用户肯定不干了，为了防止这类问题出现，我们采用下面的解决方案。

1：我们采用confirm模式+持久化+手动ack

2：消息丢失种鸽问题：消息发送失败，我们采用失败消息记录表

3：定时任务轮训失败消息记录，再次发送

这里为了防止多次重试问题，所以设置一个重试上限，并加入警告（比如一条消息最多重发5次，一旦到5次了，就给运维/开发/测试发送邮件警告）。

重复消费问题

这里是对用户积分增加，所以，绝对不能重复消费，不然这样会导致用户积分暴增，数据会出现一致性问题。解决：

1：每个消息有一个唯一的reqId，reqId=业务前缀+UUID+年月日时分秒毫秒的时间戳

2：在对比是否重复消费之前，对用户加上分布式锁，key=固定用户分布式锁前缀+userId

整体流程图

标配版：

图片

标配版

为了更好地监控消息发送失败问题，我们还可以对标配版进行升级。

图片

升级版

其他问题

我们上面说了，为了防止消息丢失，采用confirm模式+持久化+手动ack

但，实现起来并非那么简单，如果没有做过，很多东西是无法体会到的。

在使用confirm模式时，新的问题来了。

图片

问题

我们Spring中，一个Bean默认是单列的，这样的话会造成一个RabbitMQTemplate只能绑定一个confirm，这就不对了，我们需要RabbitMQTemplate不受Spring这个影响，很多人第一印象想到的就是采用原型模式。也就是在bean上添加注解：@Scope(“prototype”) 但，问题来了，比如在一个producer bean里注入RabbitMQTemplate，他最终还是认为你这个RabbitMQTemplate是单列，又和上面原型违背了，网上很多办法是给这个producer也搞成原型模式。

这个确实能解决这个问题。

说白了就是 从请求开始的bean开始到最后发送消息，这个过程的bean要都是原型模式才行。

比如：controller–service–producer

挖了个蛐蛐，问题又来了，项目中定时任务采用的是xxl-job，它的每个job都必须是单列的，上面的办法又不行了。

绝招：用Spring中的ApplicationContext**的getBean方法直接获取对应的Bean就不存在问题。

这里有点绕哈，说白了就是必须使用原型，不能使用单例。

核心代码

定义原型的rabbitTemplate。

1
2
3
4
5
6
7
8
9
10
11
12
less复制代码@Configuration
public class RabbitConfig {
    //省略部分非核心代码
    @Bean
    @Scope("prototype")
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate();
        rabbitTemplate.setConnectionFactory(connectionFactory); 
        rabbitTemplate.setMandatory(true);
        return rabbitTemplate;
    }
}

confirm模式

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
scss复制代码@Component
public class UserPointProducer { 
    @Resource
    private RetryMessageMapper chargeUserMapper;

    public void sendMessage(String message) {
        RabbitTemplate rabbitTemplate = ApplicationContextFactory.getBean(RabbitTemplate.class);
        CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            if (ack) {
                log.info("UserPointConfirm ConfirmCallback 关联数据：{},投递成功,确认情况：{}", correlationData, ack);
            } else {
                RetryMessage retryMessage = new RetryMessage();
                retryMessage.setContent(message);
                retryMessage.setRetry(5);
                retryMessage.setCreateTime(new Date());
                retryMessage.setStatus(0);
                retryMessage.setRetriedTimes(0);
                retryMessage.setType(RabbitMQConstantEnum.USER_POINT.getType());
                chargeUserMapper.insert(retryMessage);
                log.info("UserPointConfirm ConfirmCallback 关联数据：{},投递失败,确认情况：{}，原因：{}", correlationData, ack, cause);
            }
        });
 
        //后面+1 主要是为了掩饰消息发送失败
        rabbitTemplate.convertAndSend(RabbitMQConstantEnum.USER_POINT.getExchange()+"1"
                , RabbitMQConstantEnum.USER_POINT.getRoutingKey(), message, message1 -> {
                    message1.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT); // 设置消息持久化
                    return message1;
                }, correlationId);
    }
}

下面来写个测试发送案例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
typescript复制代码/**
 * {@code @description:} 测试案例
 *
 * @author tianwc 公众号：Java后端技术全栈
 * 在线刷题 1200+java面试题和1000+篇技术文章：https://woaijava.cc/
 * {@code @date:} 2024-03-24 9:19
 * {@code @version:} 1.0
 */
@GetMapping("/send")
public void send() {
    UserPointMessage userPointMessage = new UserPointMessage();
    userPointMessage.setType(UserUpdatePointEnum.ADD.getType());
    userPointMessage.setUserId(1L);
    userPointMessage.setPoint(9);
    userPointMessage.setReqId(MessageReqIdPrefixConstant.USER_POINT + UUID.randomUUID()+ DateUtils.formatDefaultDateMs());
    userPointProducer.sendMessage(JSON.toJSONString(userPointMessage));
}

消费者：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
less复制代码/**
 * {@code @description:} 用户积分消息消费者
 *
 * @author tianwc 公众号：Java后端技术全栈
 * 在线刷题 1200+java面试题和1000+篇技术文章：<a href="https://woaijava.cc/">博客地址</a>
 * {@code @date:} 2024-03-24 9:19
 * {@code @version:} 1.0
 */
@RabbitListener(queues = "user.point.queue")
@Component
@Slf4j
public class UserPointConsumer {

    @Resource
    private UserPointService userPointService;

    @RabbitHandler
    public void process(Object data, Channel channel, Message message) throws IOException {
        try {
            log.info("消费者接受到的消息是：{},消息体为：{}", data, message);
            UserPointMessage userPointMessage = JSON.parseObject(new String(message.getBody()), UserPointMessage.class);
            userPointService.updateUserPoint(userPointMessage);
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception exception) {
            channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
        }
    }
}

具体业务逻辑实现：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
less复制代码/**
 * {@code @description:} 用户积分扣除消费者服务实现类
 *
 * @author tianwc 公众号：Java后端技术全栈
 * 在线刷题 1200+java面试题和1000+篇技术文章：<a href="https://woaijava.cc/">博客地址</a>
 * {@code @date:} 2024-03-24 9:19
 * {@code @version:} 1.0
 */
@Slf4j
@Service
public class UserPointServiceImpl implements UserPointService {
    @Resource
    private ChargeUserMapper chargeUserMapper;
    @Resource
    private RedissonClient redissonClient;
    @Resource
    private PointsChangeRecordMapper pointsChangeRecordMapper;

    @Transactional(rollbackFor = Exception.class)
    @Override
    public void updateUserPoint(UserPointMessage userPointMessage) {
        RLock lock = redissonClient.getLock(RedisConstantPre.USER_INFO_ID_PRE + userPointMessage.getUserId());
        lock.lock();
        try {
            int count = pointsChangeRecordMapper.selectByReqId(userPointMessage.getReqId());
            if (count > 0) {
                log.info("消息体参数 【重复消息】:{}", userPointMessage);
                return;
            }
            PointsChangeRecord pointsChangeRecord = new PointsChangeRecord();
            pointsChangeRecord.setUserId(userPointMessage.getUserId());
            pointsChangeRecord.setPoint(userPointMessage.getPoint());
            pointsChangeRecord.setType(userPointMessage.getType());
            pointsChangeRecord.setCreateTime(new Date());
            pointsChangeRecord.setReqId(userPointMessage.getReqId());
            //积分变动记录
            pointsChangeRecordMapper.insert(pointsChangeRecord);

            ChargeUser chargeUser = chargeUserMapper.selectByPrimaryKey(userPointMessage.getUserId());
            if (userPointMessage.getType() == UserUpdatePointEnum.ADD.getType()) {
                chargeUser.setPoint(chargeUser.getPoint() + userPointMessage.getPoint());
            } else {
                chargeUser.setPoint(chargeUser.getPoint() - userPointMessage.getPoint());
            }
            //用户积分变动
            chargeUserMapper.updateByPrimaryKey(chargeUser);
        } finally {
            lock.unlock();
        }
    }
}

这里的分布式锁的好处：

  • 保证了这个重复消费部分代码的原子性
  • 保证了此时只有一个线程对用户积分进行修改

其实，正常情况下，不会走失败消息记录表，但是作为程序不得不多考虑点。

定时任务部分代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
java复制代码/**
 * {@code @description:} 用户积分消息发送job
 *
 * @author tianwc 公众号：Java后端技术全栈
 * 在线刷题 1200+java面试题和1000+篇技术文章：https://woaijava.cc/
 * {@code @date:} 2024-03-24 9:19
 * {@code @version:} 1.0
 */
@Slf4j
@Component
public class UserPointRetryMessageJob {

    @Resource
    private RetryMessageMapper retryMessageMapper;
    @Resource
    private UserPointProducer userPointProducer;

    @XxlJob("userPointRetryMessageJob")
    public void process() {
        log.info("开始执行 userPointRetryMessageJob 定时任务");
        XxlJobHelper.log("start userPointRetryMessageJob job");
        int countRetryMessage = retryMessageMapper.countRetryMessage(0, 0);
        if (countRetryMessage == 0) {
            log.info(" 执行结束 userPointRetryMessageJob 没有消息需要重发");
        }
        List<RetryMessage> retryMessages = retryMessageMapper.selectRetryMessage(0, 0);
        for (RetryMessage retryMessage : retryMessages) {
            userPointProducer.sendMessage(retryMessage);
        }
    }
}

这里还可以优化，你能想到吗？

List<RetryMessage> retryMessages = retryMessageMapper.selectRetryMessage(0, 0);

这里是一次性全部查出来了，如果出现大量消息发送失败，一次性放到本地缓存里，很容易出问题，所以，我们可以再优化成分页进行处理，比如：每次处理50条，再根据count来计算需要进行分页。

定时任务中生产者代码实现：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
scss复制代码/**
 * {@code @description:} 用户积分消息发送生产者
 *
 * @author tianwc 公众号：Java后端技术全栈
 * 在线刷题 1200+java面试题和1000+篇技术文章：https://woaijava.cc/
 * {@code @date:} 2024-03-24 9:19
 * {@code @version:} 1.0
 */
@Slf4j
@Component
public class UserPointProducer {
    @Resource
    private RetryMessageMapper chargeUserMapper;

    public void sendMessage(RetryMessage retryMessage) {
        log.info("用户积分消息重试补发,{}", retryMessage);
        String message = retryMessage.getContent();
        CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());
        RabbitTemplate rabbitTemplate = ApplicationContextFactory.getBean(RabbitTemplate.class);
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            if (ack) {
                retryMessage.setStatus(1);
                chargeUserMapper.updateByPrimaryKey(retryMessage);
                log.info("UserPointConfirm ConfirmCallback 关联数据：{},投递成功,确认情况：{}", correlationData, ack);
            } else {
                retryMessage.setRetriedTimes(retryMessage.getRetriedTimes() + 1);
                chargeUserMapper.updateByPrimaryKey(retryMessage);
                log.info("UserPointConfirm ConfirmCallback 关联数据：{},投递失败,确认情况：{}，原因：{}", correlationData, ack, cause);
            }
        });

        rabbitTemplate.convertAndSend(RabbitMQConstantEnum.USER_POINT.getExchange()
                , RabbitMQConstantEnum.USER_POINT.getRoutingKey(), message, message1 -> {
                    message1.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT); // 设置消息持久化
                    return message1;
                }, correlationId);
        log.info("用户积分消息重试补发完成");
    }
}

失败消息表

1
2
3
4
5
6
7
8
9
10
sql复制代码CREATE TABLE `retry_message` (
  `id` bigint NOT NULL AUTO_INCREMENT,
  `type` int DEFAULT NULL,
  `content` text,
  `retried_times` int DEFAULT NULL,
  `retry_limit` int NOT NULL DEFAULT '1',
  `create_time` datetime DEFAULT NULL,
  `status` int DEFAULT '0',
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8mb4 ;

一个数据库里就只要一张表即可，专门用来存储发送失败消息。

tyep：什么业务场景

content：具体消息内容

retried_times：已经重试了几次

retry_limit：重试次数限制

status：状态，是否需要重试

这里的重试次数限制，我们也可以采用配置的方式，这样就可以在分布式配置中心对此进行动态调整，不过，这个好像没什么必要，因为不会频繁地更换这个限制。直接存在表里还可以动态的针对某些业务做特殊处理，比如业务A限制次数2次，业务B次数改成3次….

没有完美的解决方法，但总有相对完美的解决方案即可。

关于积分模块，其实不止有增加积分，还有扣除积分。

比如：用户使用积分兑换优惠券，积分目前设计在用户中心，优惠券又在营销中心，所以，会涉及到分布式事务问题。

我们可以采用Seata、Atomikos、RockSeataetMQ等技术来解决，目前充电桩项目中用到过Atomikos，但是代码量实在是会增加不少，最后使用了Seata来解决分布式事务问题。

最后

希望通过本文学习，下次再遇到面试官问消息队列的两个问题，就不再是背八股文了。

好了，今天就跟大家分享这么多，希望能给你带来点点帮助。

麻烦个三连呗：点赞、转发、再看，谢谢啦！

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

0%