spring boot使用rocketmq

依賴

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
</dependency>

配置

rocketmq:
  name-server: 127.0.0.1:9876
  # 純消費者不需要以下配置
  producer:
    group: test-group

消息生產者

• 獲取客戶端模板

@Autowired
private final RocketMQTemplate rocketMQTemplate;

• 發送消息

// 默認使用同步發送, 但拿不到回執, 源碼見下文org.apache.rocketmq.spring.core.RocketMQTemplate.doSent
rocketMQTemplate.convertAndSend("test-topic", entity);
rocketMQTemplate.send("test-topic", MessageBuilder.withPayload(entity).build());
// 帶tag
rocketMQTemplate.convertAndSend("test-topic:tag1", entity);
rocketMQTemplate.send("test-topic:tag2", MessageBuilder.withPayload(entity).build());

• 發送單向消息（不關心發送結果）

rocketMQTemplate.sendOneWay("test-topic", MessageBuilder.withPayload("oneway message").build());

• 同步發送
  可選參數：delayLevel - 延遲等級
  可用延遲等級可在broker服務器配置中指定 messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h

SendResult result = rocketMQTemplate.syncSend("test-topic", entity, timeout, delayLevel);

• 發送順序消息
  可選參數：hashkey - 用於選擇消息隊列，只有在相同隊列的消息能保持順序

SendResult result = rocketMQTemplate.syncSendOrderly("test-topic", "order message", "hashkey", timeout);

• 異步發送

rocketMQTemplate.asyncSend("test-topic", MessageBuilder.withPayload(entity).build(), new SendCallback() {
    @Override
    public void onSuccess(SendResult sendResult) {}
    @Override
    public void onException(Throwable e) {}
}, timeout);

消息消費者

@Component
@RocketMQMessageListener(topic = "test-topic", consumerGroup = "test-consumer", consumeMode = ConsumeMode.ORDERLY)
public class DemoListener implements RocketMQListener<UserEntity> {
	...
    @Override
    public void onMessage(MyEntity entity) {
        logger.info("---->收到了消息了！");
        logger.info("---->" + entity.toString());
    }
}

常用RocketMQMessageListener參數

• consumeMode - 消費模式
  默認值：ConsumeMode.CONCURRENTLY並行接受
  ConsumeMode.ORDERLY每個隊列使用一個線程按順序接收
• messageModel - 消息模式
  默認值：MessageModel.CLUSTERING集群模式
  MessageModel.BROADCASTING廣播模式

重試

• 當onMessage方法拋出異常時會觸發重試，默認為無限重試

源碼

org.apache.rocketmq.spring.core.RocketMQTemplate.doSent

@Override
protected void doSend(String destination, Message<?> message) {
    SendResult sendResult = syncSend(destination, message);
    log.debug("send message to `{}` finished. result:{}", destination, sendResult);
}