# 異步消息發(fā)送失敗重試次數(shù),默認為2
rocketmq.producer.retry-times-when-send-async-failed=2
# 消息發(fā)送失敗重試次數(shù),默認為2
rocketmq.producer.retry-times-when-send-failed=2

DefaultMQProducer defaultMQProducer = new DefaultMQProducer();
defaultMQProducer.setRetryTimesWhenSendFailed(2);
defaultMQProducer.setRetryTimesWhenSendAsyncFailed(2);

rocketmq.name-server=localhost:9876
# 消費者組
rocketmq.producer.group=producer_group
rocketmq.consumer.topic=consumer_topic
rocketmq.consumer.group=consumer_group

@Component
public class RetryConsumerDemo {
    @Value("${rocketmq.name-server}")
    private String namesrvAddr;
    @Value("${rocketmq.consumer.topic}")
    private String topic;
    @Value("${rocketmq.consumer.group}")
    private String consumerGroup;
    private final DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("consumer_group");
    @PostConstruct
    public void start() {
        try {
            consumer.setNamesrvAddr(namesrvAddr);
            //設置集群消費模式
            consumer.setMessageModel(MessageModel.CLUSTERING);
            //設置消費超時時間(分鐘)
            consumer.setConsumeTimeout(1);
            //訂閱主題
            consumer.subscribe(topic , "*");
            //注冊消息監(jiān)聽器
            consumer.registerMessageListener(new MessageListenerConcurrentlyImpl());
            //最大重試次數(shù)
            consumer.setMaxReconsumeTimes(2);
            //啟動消費端
            consumer.start();
            System.out.println("Retry Consumer Start...");
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }
}

public class MessageListenerConcurrentlyImpl implements MessageListenerConcurrently {
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        if (CollectionUtils.isEmpty(msgs)) {
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
        MessageExt message = msgs.get(0);
        try {
            final LocalDateTime now = LocalDateTime.now();
            //逐條消費
            String messageBody = new String(message.getBody(), StandardCharsets.UTF_8);
            System.out.println("當前時間："+now+"， messageId: " + message.getMsgId() + ",topic: " +
                    message.getTopic()  + ",messageBody: " + messageBody);
            //模擬消費失敗
            if ("Concurrently_test".equals(messageBody)) {
                int a = 1 / 0;
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        } catch (Exception e) {
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
    }
}

//注冊消息監(jiān)聽器
consumer.registerMessageListener(new MessageListenerConcurrentlyImpl());

@RunWith(SpringRunner.class)
@SpringBootTest(classes = RocketmqApplication.class)
class RocketmqApplicationTests {
    @Value("${rocketmq.consumer.topic}")
    private String topic;
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    @Test
    public void testProducer(){
        String msg = "Concurrently_test";
        rocketMQTemplate.convertAndSend(topic , msg);
    }
}

public enum ConsumeConcurrentlyStatus {
    /**
     * Success consumption
     */
    CONSUME_SUCCESS,
    /**
     * Failure consumption,later try to consume
     */
    RECONSUME_LATER;
}

public class MessageListenerOrderlyImpl implements MessageListenerOrderly {
    @Override
    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
        if (CollectionUtils.isEmpty(msgs)) {
            return ConsumeOrderlyStatus.SUCCESS;
        }
        MessageExt message = msgs.get(0);
        try {
            final LocalDateTime now = LocalDateTime.now();
            //逐條消費
            String messageBody = new String(message.getBody(), StandardCharsets.UTF_8);
            System.out.println("當前時間："+now+"， messageId: " + message.getMsgId() + ",topic: " +
                    message.getTopic()  + ",messageBody: " + messageBody);
            //模擬消費失敗
            if ("Orderly_test".equals(messageBody)) {
                int a = 1 / 0;
            }
            return ConsumeOrderlyStatus.SUCCESS;
        } catch (Exception e) {
            return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
        }
    }
}

//最大重試次數(shù)
consumer.setMaxReconsumeTimes(2);
//順序消費 重試時間間隔
consumer.setSuspendCurrentQueueTimeMillis(2000);

@RunWith(SpringRunner.class)
@SpringBootTest(classes = RocketmqApplication.class)
class RocketmqApplicationTests {
    @Value("${rocketmq.consumer.topic}")
    private String topic;
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    @Test
    public void testProducer(){
        String msg = "Orderly_test";
        rocketMQTemplate.convertAndSend(topic , msg);
    }
}

public enum ConsumeOrderlyStatus {
    /**
     * Success consumption
     */
    SUCCESS,
    /**
     * Rollback consumption(only for binlog consumption)
     */
    @Deprecated
    ROLLBACK,
    /**
     * Commit offset(only for binlog consumption)
     */
    @Deprecated
    COMMIT,
    /**
     * Suspend current queue a moment
     */
    SUSPEND_CURRENT_QUEUE_A_MOMENT;
}