Java使用RabbitMQ保证消息幂等性的方法步骤

public String generateMessageId(String messageBody) {
    return DigestUtils.md5DigestAsHex(messageBody.getBytes(StandardCharsets.UTF_8));
}

@Autowired
private StringRedisTemplate redisTemplate;

public boolean isDuplicateMessage(String messageId) {
    Boolean exists = redisTemplate.hasKey(messageId);
    return Boolean.TRUE.equals(exists);
}

public void markMessageAsProcessed(String messageId) {
    redisTemplate.opsForValue().set(messageId, "processed", 1, TimeUnit.DAYS);
}

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.support.MessageProperties;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;

@Service
public class RabbitMQConsumer {

    @Autowired
    private StringRedisTemplate redisTemplate;

    @RabbitListener(queues = "task_queue")
    public void receiveMessage(String message, MessageProperties messageProperties) {
        String messageId = messageProperties.getMessageId();
        
        // 如果消息没有ID，则自行生成一个ID（假设消息内容不变）
        if (messageId == null || messageId.isEmpty()) {
            messageId = generateMessageId(message);
        }

        // 检查消息是否已经处理过
        if (isDuplicateMessage(messageId)) {
            System.out.println("Message with ID " + messageId + " already processed, skipping.");
            return;
        }

        try {
            // 处理消息的业务逻辑
            processMessage(message);

            // 处理成功后，标记消息ID
            markMessageAsProcessed(messageId);

        } catch (Exception e) {
            System.err.println("Failed to process message with ID " + messageId + ": " + e.getMessage());
            // 如果处理失败，可以根据业务需求选择是否重新投递消息
        }
    }

    private void processMessage(String message) {
        // 具体的消息处理逻辑
        System.out.println("Processing message: " + message);
        // 假设处理逻辑是幂等的
    }

    private String generateMessageId(String messageBody) {
        return DigestUtils.md5DigestAsHex(messageBody.getBytes(StandardCharsets.UTF_8));
    }

    private boolean isDuplicateMessage(String messageId) {
        Boolean exists = redisTemplate.hasKey(messageId);
        return Boolean.TRUE.equals(exists);
    }

    private void markMessageAsProcessed(String messageId) {
        redisTemplate.opsForValue().set(messageId, "processed", 1, TimeUnit.DAYS);
    }
}