RocketMQ重试机制和消息幂等(3)

//发送消息
            for (int i=1;i<=1;i++){
                //模拟网络延迟，每秒发送一次MQ
                Thread.sleep(1000);
                //创建消息，topic主题名称  tags临时值代表小分类， body代表消息体
                Message message=new Message("itmayiedu-topic03","TagA",("itmayiedu-"+i).getBytes());
                //消息的唯一标识
                message.setKeys("订单消息："+i);
                //发送消息
                SendResult sendResult=producer.send(message);
                System.out.println("信息幂等问题来了："+sendResult.toString());
            }
        producer.shutdown();
    }
}

4、消费者

package com.zn.idempotent;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.message.MessageExt;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.logging.LogManager;
import java.util.logging.Logger;

/**
 * 消息幂等消费者
 */
public class IdempotentConsumer {

static private Map<String, Object> logMap = new HashMap<>();

public static void main(String[] args) throws MQClientException {
        //创建消费者
        DefaultMQPushConsumer consumer=new DefaultMQPushConsumer("rmq-group");
        //设置NameServer地址
        consumer.setNamesrvAddr("192.168.33.135:9876;192.168.33.136:9876");
        //设置实例名称
        consumer.setInstanceName("consumer");
        //订阅topic
        consumer.subscribe("itmayiedu-topic03","TagA");

//监听消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                String key=null;
                String msgId=null;

for (MessageExt messageExt:list){
                        key=messageExt.getKeys();
                        //判读redis中有没有当前消息key
                        if (logMap.containsKey(key)) {
                            // 无需继续重试。
                            System.out.println("key:"+key+",已经消费，无需重试...");
                            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                        }
                        //RocketMQ由于是集群环境，所以产生的消息ID可能会重复
                        msgId = messageExt.getMsgId();
                        System.out.println("key:" + key + ",msgid:" + msgId + "---" + new String(messageExt.getBody()));
                        //将当前key保存在redis中
                        logMap.put(messageExt.getKeys(),messageExt);
                    }
                try {
                    int i=5/0;
                }catch (Exception e){
                    e.printStackTrace();
                    //人工补偿
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
    });
        //启动消费者
        consumer.start();
        System.out.println("Consumer Started!");
    }
}

5、控制台效果