@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;

    /**
     * 通用的發(fā)送方法，允許自定義消息內(nèi)容和處理邏輯
     * @param messagePayload 消息體數(shù)據(jù)
     */
    @Override
    public void send(Map<String, String> messagePayload) {
        // 生成唯一消息鍵，用于標(biāo)識(shí)消息的冪等性
        String messageId = UUID.randomUUID().toString();
        messagePayload.put("messageId", messageId);

        // 構(gòu)建消息
        Message<Map<String, String>> message = MessageBuilder
                .withPayload(messagePayload)
                .setHeader(MessageConst.PROPERTY_KEYS, messageId)
                .build();

        // 發(fā)送消息并處理結(jié)果
        try {
            SendResult sendResult = rocketMQTemplate.syncSend(customTopic, message, 2000L);
            log.info("消息發(fā)送成功: 狀態(tài)={}, 消息ID={}, 消息鍵={}", sendResult.getSendStatus(), sendResult.getMsgId(), messageId);
        } catch (Exception e) {
            log.error("消息發(fā)送失敗: 消息內(nèi)容={}", 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 消息唯一標(biāo)識(shí)
     * @return true 表示消息未處理，可以繼續(xù)處理；false 表示消息已處理，避免重復(fù)消費(fèi)
     */
    public boolean isMessageNotProcessed(String messageId) {
        String key = IDEMPOTENT_KEY_PREFIX + messageId;
        // 嘗試設(shè)置新鍵，如果不存在則返回 true 表示未處理，存在則返回 false
        return Boolean.TRUE.equals(stringRedisTemplate.opsForValue().setIfAbsent(key, "0", 2, TimeUnit.MINUTES));
    }

    /**
     * 標(biāo)記消息處理流程完成
     *
     * @param messageId 消息唯一標(biāo)識(shí)
     */
    public void markAsProcessed(String messageId) {
        String key = IDEMPOTENT_KEY_PREFIX + messageId;
        stringRedisTemplate.opsForValue().set(key, "1", 2, TimeUnit.MINUTES);
    }

    /**
     * 查詢消息是否已經(jīng)處理完成
     *
     * @param messageId 消息唯一標(biāo)識(shí)
     * @return true 表示消息處理已完成，false 表示未完成
     */
    public boolean isProcessingComplete(String messageId) {
        String key = IDEMPOTENT_KEY_PREFIX + messageId;
        return Objects.equals(stringRedisTemplate.opsForValue().get(key), "1");
    }

    /**
     * 處理異常時(shí)刪除冪等標(biāo)識(shí)
     *
     * @param messageId 消息唯一標(biāo)識(shí)
     */
    public void clearProcessedFlag(String messageId) {
        String key = IDEMPOTENT_KEY_PREFIX + messageId;
        stringRedisTemplate.delete(key);
    }
}

@Override
public void onMessage(Map<String, String> producerMap) {
    // 獲取消息的唯一標(biāo)識(shí)符
    String keys = producerMap.get("keys");
    
    // 檢查是否已處理過該消息，冪等性控制
    if (!messageQueueIdempotentHandler.isMessageProcessed(keys)) {
        // 若該消息流程尚未完成
        if (messageQueueIdempotentHandler.isAccomplish(keys)) {
            return; // 跳過已完成流程的消息
        }
        throw new ServiceException("消息未完成流程，需要消息隊(duì)列重試");
    }
    
    // 業(yè)務(wù)邏輯處理
    try {
       	//調(diào)用業(yè)務(wù)方法代碼
        }
    } catch (Throwable ex) {
        log.error("消費(fèi)異常", ex);
        throw ex;
    }
    
    // 標(biāo)記消息處理完成
    messageQueueIdempotentHandler.setAccomplish(keys);
}