rocketmq 消息幂等

RocketMQ消息幂等是指在消息重复消费时，不会对业务数据产生重复的影响。通常情况下，消息重复消费是由于网络抖动、消息重发等原因导致的。为了避免这种情况，可以采用以下两种方式实现RocketMQ消息幂等：

1. 利用业务主键去重：在消息发送时，将业务主键作为消息的唯一标识，消费者在消费消息时，先查询该业务主键是否已经存在，如果已经存在，则不再进行处理，否则进行业务处理。

2. 利用消息ID去重：在消息发送时，将消息ID作为消息的唯一标识，消费者在消费消息时，先查询该消息ID是否已经存在，如果已经存在，则不再进行处理，否则进行业务处理。

为了更好地实现RocketMQ消息幂等，可以使用通用的RocketMQ消息幂等去重消费者工具类，该工具类支持利用Redis或者MySQL做幂等表，支持业务主键去重或消息ID去重，同时还支持消息重复并发控制，确保重复的消息在消费成功/失败前不会同时被消费第二次。

相关问题

rocketmq保证幂等性 用redistributionTemplate写一个代码示例

好的，以下是使用 `DefaultMQProducer` 和 `TransactionMQProducer` 对消息进行幂等性保障的代码示例。

1. 使用 `DefaultMQProducer`

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageConst;

public class IdempotentProducer {

    private static final String TOPIC_NAME = "test_topic";
    private static final String TAG_NAME = "test_tag";
    private static final String PRODUCER_GROUP_NAME = "test_group";

    public static void main(String[] args) throws Exception {
        // 实例化消息生产者
        DefaultMQProducer producer = new DefaultMQProducer(PRODUCER_GROUP_NAME);
        producer.setNamesrvAddr("localhost:9876");
        // 启动消息生产者
        producer.start();

        try {
            // 创建消息
            Message message = new Message(TOPIC_NAME, TAG_NAME, "Hello world!".getBytes());
            // 设置消息的唯一键，用于幂等性保障
            message.putUserProperty(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX, "123456");

            // 发送消息
            producer.send(message);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 关闭消息生产者
            producer.shutdown();
        }
    }
}

2. 使用 `TransactionMQProducer`

import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.common.protocol.header.CheckTransactionStateRequestHeader;

import java.util.List;
import java.util.concurrent.*;

public class IdempotentTransactionProducer {

    private static final String TOPIC_NAME = "test_topic";
    private static final String TAG_NAME = "test_tag";
    private static final String PRODUCER_GROUP_NAME = "test_group";

    public static void main(String[] args) throws Exception {
        // 实例化事务消息生产者
        TransactionMQProducer producer = new TransactionMQProducer(PRODUCER_GROUP_NAME);
        producer.setNamesrvAddr("localhost:9876");
        // 设置事务监听器
        producer.setTransactionListener(new TransactionListenerImpl());
        // 启动事务消息生产者
        producer.start();

        try {
            // 创建消息
            Message message = new Message(TOPIC_NAME, TAG_NAME, "Hello world!".getBytes());

            // 发送事务消息
            producer.sendMessageInTransaction(message, null);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 关闭事务消息生产者
            producer.shutdown();
        }
    }

    /**
     * 事务监听器实现类
     */
    static class TransactionListenerImpl implements TransactionMQProducer.TransactionListener {
        private final ConcurrentHashMap<String, Integer> localTrans = new ConcurrentHashMap<>();
        private final ExecutorService executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(2000), r -> {
            Thread thread = new Thread(r);
            thread.setName("client-transaction-msg-check-thread");
            return thread;
        });

        /**
         * 执行本地事务
         */
        @Override
        public LocalTransactionState executeLocalTransaction(Message message, Object arg) {
            // 执行本地事务逻辑
            // ...

            // 记录本地事务状态
            localTrans.put(message.getTransactionId(), 1);

            // 返回本地事务状态
            return LocalTransactionState.UNKNOW;
        }

        /**
         * 检查事务状态
         */
        @Override
        public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
            // 获取本地事务状态
            Integer status = localTrans.get(messageExt.getTransactionId());

            // 如果本地事务状态为 COMMIT_MESSAGE，则返回 COMMIT_MESSAGE
            if (status != null) {
                switch (status) {
                    case 0:
                        return LocalTransactionState.UNKNOW;
                    case 1:
                        return LocalTransactionState.COMMIT_MESSAGE;
                    case 2:
                        return LocalTransactionState.ROLLBACK_MESSAGE;
                }
            }

            // 如果本地事务状态未知，则发送事务状态检查请求
            MessageQueue queue = new MessageQueue();
            queue.setBrokerName(messageExt.getBrokerName());
            queue.setQueueId(messageExt.getQueueId());
            queue.setTopic(TOPIC_NAME);
            CheckTransactionStateRequestHeader requestHeader = new CheckTransactionStateRequestHeader();
            requestHeader.setCommitLogOffset(messageExt.getCommitLogOffset());
            requestHeader.setMsgId(messageExt.getMsgId());
            requestHeader.setTransactionId(messageExt.getTransactionId());
            requestHeader.setTranStateTableOffset(messageExt.getQueueOffset());
            try {
                CompletableFuture<LocalTransactionState> future = new CompletableFuture<>();
                executorService.submit(() -> {
                    try {
                        // 发送事务状态检查请求，并等待响应结果
                        LocalTransactionState state = producer.checkTransactionState(requestHeader, queue);
                        future.complete(state);
                    } catch (Exception e) {
                        future.complete(LocalTransactionState.UNKNOW);
                    }
                });
                // 设置事务状态检查超时时间
                return future.get(5, TimeUnit.SECONDS);
            } catch (TimeoutException e) {
                return LocalTransactionState.UNKNOW;
            } catch (InterruptedException e) {
                return LocalTransactionState.UNKNOW;
            } catch (ExecutionException e) {
                return LocalTransactionState.UNKNOW;
            }
        }
    }
}

rocketmq幂等

RocketMQ 幂等是指在消息的生产和消费过程中，保证消息处理的幂等性。幂等性是指对同一操作的多次执行所产生的效果与一次执行的效果相同。

在 RocketMQ 中，我们可以通过以下方式实现消息的幂等性：

1. 唯一标识：在消息体中添加一个唯一的标识符，每次处理消息时先判断是否已经处理过该标识符对应的消息，如果已经处理过，则直接忽略。

2. 重复消费检测：RocketMQ 提供了消息消费状态存储机制，在消费端可以记录消费状态，当接收到重复消息时，可以判断是否已经处理过。

3. 幂等消息处理逻辑：通过在业务逻辑中增加幂等性处理，确保同一条消息被处理多次时只产生一次结果。例如，对于数据库操作，可以使用唯一索引来保证数据的唯一性，在插入前先查询是否存在相同数据。

4. 分布式锁：在多个消费者并发处理消息时，可以使用分布式锁来保证只有一个消费者能够处理该消息，从而避免重复处理。

以上是常用的几种实现 RocketMQ 幂等性的方法，根据具体的业务场景和需求，可以选择适合的方式来实现幂等性保证。