RabbitMq实战如何保证消息幂等消费

2022-05-27 539

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

本文涉及的产品

Redis 开源版，标准版 2GB

云数据库 Tair（兼容Redis），内存型 2GB

简介： RabbitMq实战如何保证消息幂等消费

正文

一、消息幂等性

在编程中一个幂等操作的特点是其任意多次执行所产生的结果与一次执行的产生的结果相同，在mq中由于网络故障或客户端延迟消费mq自动重试过程中可能会导致消息的重复消费，那我们如何保证消息的幂等问题呢？也可以理解为如何保证消息不被重复消费呢，不重复消费也就解决了幂等问题。

二、解决方案

1、生成全局id，存入redis或者数据库，在消费者消费消息之前，查询一下该消息是否有消费过。

2、如果该消息已经消费过，则告诉mq消息已经消费，将该消息丢弃（手动ack）。

3、如果没有消费过，将该消息进行消费并将消费记录写进redis或者数据库中。

注：还有一种方式，数据库操作可以设置唯一键（消息id），防止重复数据的插入，这样插入只会报错而不会插入重复数据，本人没有测试。

三、代码

简单描述一下需求，如果订单完成之后，需要为用户累加积分，又需要保证积分不会重复累加。那么再mq消费消息之前，先去数据库查询该消息是否已经消费，如果已经消费那么直接丢弃消息。

如果是Redis存放数据key=全局id，value=积分值，在消费消息之前，通过全局id去redis查询是否有该数据，如果有直接丢弃。该方法本人没有测试，只是说说自己的思路。有不对的希望大佬们不吝赐教。

生产者

package com.xiaojie.score.producer;
import com.alibaba.fastjson.JSONObject;
import com.xiaojie.score.entity.Score;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import java.util.UUID;
/**
 * @author xiaojie
 * @version 1.0
 * @description:发送积分消息的生产者
 * @date 2021/10/10 22:18
 */
@Component
@Slf4j
public class ScoreProducer implements RabbitTemplate.ConfirmCallback {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    //定义交换机
    private static final String SCORE_EXCHANGE = "xiaojie_score_exchaneg";
    //定义路由键
    private static final String SCORE_ROUTINNGKEY = "score.add";
    /**
     * @description: 订单完成
     * @param:
     * @return: java.lang.String
     * @author xiaojie
     * @date: 2021/10/10 22:30
     */
    public String completeOrder() {
        String orderId = UUID.randomUUID().toString();
        System.out.println("订单已完成");
        //发送积分通知
        Score score = new Score();
        score.setScore(100);
        score.setOrderId(orderId);
        String jsonMSg = JSONObject.toJSONString(score);
        sendScoreMsg(jsonMSg, orderId);
        return orderId;
    }
    /**
     * @description: 发送积分消息
     * @param:
     * @param: message
     * @param: orderId
     * @return: void
     * @author xiaojie
     * @date: 2021/10/10 22:22
     */
    @Async
    public void sendScoreMsg(String jsonMSg, String orderId) {
        this.rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.convertAndSend(SCORE_EXCHANGE, SCORE_ROUTINNGKEY, jsonMSg, message -> {
            //设置消息的id为唯一
            message.getMessageProperties().setMessageId(orderId);
            return message;
        });
    }
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String s) {
        if (ack) {
            log.info(">>>>>>>>消息发送成功:correlationData:{},ack:{},s:{}", correlationData, ack, s);
        } else {
            log.info(">>>>>>>消息发送失败{}", ack);
        }
    }
}

消费者

package com.xiaojie.score.consumer;
import com.alibaba.fastjson.JSONObject;
import com.rabbitmq.client.Channel;
import com.xiaojie.score.entity.Score;
import com.xiaojie.score.mapper.ScoreMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.stereotype.Component;
import java.io.IOException;
import java.util.Map;
/**
 * @author xiaojie
 * @version 1.0
 * @description: 积分的消费者
 * @date 2021/10/10 22:37
 */
@Component
@Slf4j
public class ScoreConsumer {
    @Autowired
    private ScoreMapper scoreMapper;
    @RabbitListener(queues = {"xiaojie_score_queue"})
    public void onMessage(Message message, @Headers Map<String, Object> headers, Channel channel) throws IOException {
        String orderId = message.getMessageProperties().getMessageId();
        if (StringUtils.isBlank(orderId)) {
            return;
        }
        log.info(">>>>>>>>消息id是：{}", orderId);
        String msg = new String(message.getBody());
        Score score = JSONObject.parseObject(msg, Score.class);
        if (score == null) {
            return;
        }
        //执行前去数据库查询，是否存在该数据，存在说明已经消费成功，不存在就去添加数据，添加成功丢弃消息
        Score dbScore = scoreMapper.selectByOrderId(orderId);
        if (dbScore != null) {
            //证明已经消费消息，告诉mq已经消费,丢弃消息
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            return;
        }
        Integer result = scoreMapper.save(score);
        if (result > 0) {
            //积分已经累加，删除消息
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            return;
        } else {
            log.info("消费失败，采取相应的人工补偿");
        }
    }
}