Redis处理MQ消费幂等的实现示例

@slf4j
@component
@requiredargsconstructor
@conditionalonproperty(name = "message-queue.type", havingvalue = "rocketmq")
public class custommessageproducer implements messagequeueproducer {
    
    private final rocketmqtemplate rocketmqtemplate;

    @value("${rocketmq.producer.topic}")
    private string customtopic;

    /**
     * 通用的发送方法，允许自定义消息内容和处理逻辑
     * @param messagepayload 消息体数据
     */
    @override
    public void send(map<string, string> messagepayload) {
        // 生成唯一消息键，用于标识消息的幂等性
        string messageid = uuid.randomuuid().tostring();
        messagepayload.put("messageid", messageid);

        // 构建消息
        message<map<string, string>> message = messagebuilder
                .withpayload(messagepayload)
                .setheader(messageconst.property_keys, messageid)
                .build();

        // 发送消息并处理结果
        try {
            sendresult sendresult = rocketmqtemplate.syncsend(customtopic, message, 2000l);
            log.info("消息发送成功: 状态={}, 消息id={}, 消息键={}", sendresult.getsendstatus(), sendresult.getmsgid(), messageid);
        } catch (exception e) {
            log.error("消息发送失败: 消息内容={}", json.tojsonstring(messagepayload), e);
            // 添加自定义的失败处理逻辑
        }
    }
}

package com.example.project.mq.idempotent;

import lombok.requiredargsconstructor;
import org.springframework.data.redis.core.stringredistemplate;
import org.springframework.stereotype.component;

import java.util.objects;
import java.util.concurrent.timeunit;

/**
 * 消息幂等处理器
 */
@component
@requiredargsconstructor
public class messageidempotenthandler {

    private final stringredistemplate stringredistemplate;
    private static final string idempotent_key_prefix = "message:idempotent:";

    /**
     * 判断消息是否已被处理
     *
     * @param messageid 消息唯一标识
     * @return true 表示消息未处理，可以继续处理；false 表示消息已处理，避免重复消费
     */
    public boolean ismessagenotprocessed(string messageid) {
        string key = idempotent_key_prefix + messageid;
        // 尝试设置新键，如果不存在则返回 true 表示未处理，存在则返回 false
        return boolean.true.equals(stringredistemplate.opsforvalue().setifabsent(key, "0", 2, timeunit.minutes));
    }

    /**
     * 标记消息处理流程完成
     *
     * @param messageid 消息唯一标识
     */
    public void markasprocessed(string messageid) {
        string key = idempotent_key_prefix + messageid;
        stringredistemplate.opsforvalue().set(key, "1", 2, timeunit.minutes);
    }

    /**
     * 查询消息是否已经处理完成
     *
     * @param messageid 消息唯一标识
     * @return true 表示消息处理已完成，false 表示未完成
     */
    public boolean isprocessingcomplete(string messageid) {
        string key = idempotent_key_prefix + messageid;
        return objects.equals(stringredistemplate.opsforvalue().get(key), "1");
    }

    /**
     * 处理异常时删除幂等标识
     *
     * @param messageid 消息唯一标识
     */
    public void clearprocessedflag(string messageid) {
        string key = idempotent_key_prefix + messageid;
        stringredistemplate.delete(key);
    }
}

@override
public void onmessage(map<string, string> producermap) {
    // 获取消息的唯一标识符
    string keys = producermap.get("keys");
    
    // 检查是否已处理过该消息，幂等性控制
    if (!messagequeueidempotenthandler.ismessageprocessed(keys)) {
        // 若该消息流程尚未完成
        if (messagequeueidempotenthandler.isaccomplish(keys)) {
            return; // 跳过已完成流程的消息
        }
        throw new serviceexception("消息未完成流程，需要消息队列重试");
    }
    
    // 业务逻辑处理
    try {
       	//调用业务方法代码
        }
    } catch (throwable ex) {
        log.error("消费异常", ex);
        throw ex;
    }
    
    // 标记消息处理完成
    messagequeueidempotenthandler.setaccomplish(keys);
}