rocketmq-spring-boot

English

帮助开发者在Spring Boot中快速集成RocketMQ。支持Spring Message规范，方便开发者从其它MQ快速切换到RocketMQ。

如何贡献和帮助社区

我们永远欢迎开发者的帮助来使这个项目更加完善，无论是小的文档还是大的功能新特性，请参考RocketMQ的主站了解细节

前提条件

• JDK 1.8 and above
• Maven 3.0 and above

功能特性：

• [x] 同步发送
• [x] 同步顺序发送
• [x] 异步发送
• [x] 异步顺序发送
• [x] 顺序消费
• [x] 并发消费（广播/集群）
• [x] one-way方式发送
• [x] 事务方式发送
• [ ] pull消费

Quick Start

下面列出来了一些关键点，完整的示例请参考： rocketmq-spring-boot-samples

注意:当前的RELEASE.VERSION=2.0.0

<!--在pom.xml中添加依赖-->
<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <version>${RELEASE.VERSION}</version>
</dependency>

发送消息

## application.properties
spring.rocketmq.nameServer=127.0.0.1:9876
spring.rocketmq.producer.group=my-group

注意:

请将上述示例配置中的127.0.0.1:9876替换成真实RocketMQ的NameServer地址与端口

@SpringBootApplication
public class ProducerApplication implements CommandLineRunner{
    @Resource
    private RocketMQTemplate rocketMQTemplate;
    
    public static void main(String[] args){
        SpringApplication.run(ProducerApplication.class, args);
    }
    
    public void run(String... args) throws Exception {
        rocketMQTemplate.convertAndSend("test-topic-1", "Hello, World!");
        rocketMQTemplate.send("test-topic-1", MessageBuilder.withPayload("Hello, World! I'm from spring message").build());
        rocketMQTemplate.convertAndSend("test-topic-2", new OrderPaidEvent("T_001", new BigDecimal("88.00")));
        
//        rocketMQTemplate.destroy(); // notes:  once rocketMQTemplate be destroyed, you can not send any message again with this rocketMQTemplate
    }
    
    @Data
    @AllArgsConstructor
    public class OrderPaidEvent implements Serializable{
        private String orderId;
        
        private BigDecimal paidMoney;
    }
}

在发送客户端发送事务性消息并且实现回查Listener

@SpringBootApplication
public class ProducerApplication implements CommandLineRunner{
    @Resource
    private RocketMQTemplate rocketMQTemplate;

    public static void main(String[] args){
        SpringApplication.run(ProducerApplication.class, args);
    }

    public void run(String... args) throws Exception {
        try {
            // Build a SpringMessage for sending in transaction
            Message msg = MessageBuilder.withPayload(..)...
            // In sendMessageInTransaction(), the first parameter transaction name ("test")
            // must be same with the @RocketMQTransactionListener's member field 'transName'
            rocketMQTemplate.sendMessageInTransaction("test", "test-topic" msg, null);
        } catch (MQClientException e) {
            e.printStackTrace(System.out);
        }
    }

    // Define transaction listener with the annotation @RocketMQTransactionListener
    @RocketMQTransactionListener(transName="test")
    class TransactionListenerImpl implements RocketMQLocalTransactionListener {
          @Override
          public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
            // ... local transaction process, return bollback, commit or unknown
            return RocketMQLocalTransactionState.UNKNOWN;
          }

          @Override
          public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
            // ... check transaction status and return bollback, commit or unknown
            return RocketMQLocalTransactionState.COMMIT;
          }
    }
}

更多发送相关配置

spring.rocketmq.producer.retryTimesWhenSendAsyncFailed=0
spring.rocketmq.producer.sendMessageTimeout=300000
spring.rocketmq.producer.compressMessageBodyOverHowmuch=4096
spring.rocketmq.producer.maxMessageSize=4194304
spring.rocketmq.producer.retryAnotherBrokerWhenNotStoreOk=false
spring.rocketmq.producer.retryTimesWhenSendFailed=2

接收消息

## application.properties
spring.rocketmq.nameServer=127.0.0.1:9876

注意:

请将上述示例配置中的127.0.0.1:9876替换成真实RocketMQ的NameServer地址与端口

@SpringBootApplication
public class ConsumerApplication{
    
    public static void main(String[] args){
        SpringApplication.run(ConsumerApplication.class, args);
    }
    
    @Slf4j
    @Service
    @RocketMQMessageListener(topic = "test-topic-1", consumerGroup = "my-consumer_test-topic-1")
    public class MyConsumer1 implements RocketMQListener<String>{
        public void onMessage(String message) {
            log.info("received message: {}", message);
        }
    }
    
    @Slf4j
    @Service
    @RocketMQMessageListener(topic = "test-topic-2", consumerGroup = "my-consumer_test-topic-2")
    public class MyConsumer2 implements RocketMQListener<OrderPaidEvent>{
        public void onMessage(OrderPaidEvent orderPaidEvent) {
            log.info("received orderPaidEvent: {}", orderPaidEvent);
        }
    }
}

更多消费相关配置

see: RocketMQMessageListener

FAQ

1. 生产环境有多个nameserver该如何连接？

   spring.rocketmq.nameServer支持配置多个nameserver地址，采用;分隔即可。例如：172.19.0.1:9876;172.19.0.2:9876

2. rocketMQTemplate在什么时候被销毁？

   开发者在项目中使用rocketMQTemplate发送消息时，不需要手动执行rocketMQTemplate.destroy()方法， rocketMQTemplate会在spring容器销毁时自动销毁。

3. 启动报错：Caused by: org.apache.rocketmq.client.exception.MQClientException: The consumer group[xxx] has been created before, specify another name please

   RocketMQ在设计时就不希望一个消费者同时处理多个类型的消息，因此同一个consumerGroup下的consumer职责应该是一样的，不要干不同的事情（即消费多个topic）。建议consumerGroup与topic一一对应。

4. 发送的消息内容体是如何被序列化与反序列化的？

   RocketMQ的消息体都是以byte[]方式存储。当业务系统的消息内容体如果是java.lang.String类型时，统一按照utf-8编码转成byte[]；如果业务系统的消息内容为非java.lang.String类型，则采用jackson-databind序列化成JSON格式的字符串之后，再统一按照utf-8编码转成byte[]。

5. 如何指定topic的tags?

   RocketMQ的最佳实践中推荐：一个应用尽可能用一个Topic，消息子类型用tags来标识，tags可以由应用自由设置。 在使用rocketMQTemplate发送消息时，通过设置发送方法的destination参数来设置消息的目的地，destination的格式为topicName:tagName，:前面表示topic的名称，后面表示tags名称。

   注意:

   tags从命名来看像是一个复数，但发送消息时，目的地只能指定一个topic下的一个tag，不能指定多个。

6. 发送消息时如何设置消息的key?

   可以通过重载的xxxSend(String destination, Message<?> msg, ...)方法来发送消息，指定msg的headers来完成。示例：

   Message<?> message = MessageBuilder.withPayload(payload).setHeader(MessageConst.PROPERTY_KEYS, msgId).build();
   rocketMQTemplate.send("topic-test", message);
   

   同理还可以根据上面的方式来设置消息的FLAG、WAIT_STORE_MSG_OK以及一些用户自定义的其它头信息。

   注意:

   在将Spring的Message转化为RocketMQ的Message时，为防止header信息与RocketMQ的系统属性冲突，在所有header的名称前面都统一添加了前缀USERS_。因此在消费时如果想获取自定义的消息头信息，请遍历头信息中以USERS_开头的key即可。

7. 消费消息时，除了获取消息payload外，还想获取RocketMQ消息的其它系统属性，需要怎么做？

   消费者在实现RocketMQListener接口时，只需要起泛型为MessageExt即可，这样在onMessage方法将接收到RocketMQ原生的MessageExt消息。

   @Slf4j
   @Service
   @RocketMQMessageListener(topic = "test-topic-1", consumerGroup = "my-consumer_test-topic-1")
   public class MyConsumer2 implements RocketMQListener<MessageExt>{
       public void onMessage(MessageExt messageExt) {
           log.info("received messageExt: {}", messageExt);
       }
   }
   

8. 如何指定消费者从哪开始消费消息，或开始消费的位置？

   消费者默认开始消费的位置请参考：RocketMQ FAQ。 若想自定义消费者开始的消费位置，只需在消费者类添加一个RocketMQPushConsumerLifecycleListener接口的实现即可。 示例如下：

   @Slf4j
   @Service
   @RocketMQMessageListener(topic = "test-topic-1", consumerGroup = "my-consumer_test-topic-1")
   public class MyConsumer1 implements RocketMQListener<String>, RocketMQPushConsumerLifecycleListener {
       @Override
       public void onMessage(String message) {
           log.info("received message: {}", message);
       }
   
       @Override
       public void prepareStart(final DefaultMQPushConsumer consumer) {
           // set consumer consume message from now
           consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_TIMESTAMP);
           consumer.setConsumeTimestamp(UtilAll.timeMillisToHumanString3(System.currentTimeMillis()));
       }
   }
   

   同理，任何关于DefaultMQPushConsumer的更多其它其它配置，都可以采用上述方式来完成。

9. 如何发送事务消息(即半消息支持分布式事务)？ 在客户端，首先用户需要实现RocketMQLocalTransactionListener接口，并在接口类上注解声明@RocketMQTransactionListener，实现确认和回查方法；然后再使用资源模板RocketMQTemplate， 调用方法sendMessageInTransaction()来进行消息的发布。 注意：这个方法通过指定发送者组名与具体的声明了txProducerGroup的TransactionListener进行关联，您也可以不指定这个值，从而使用默认的事务发送者组。