spring-boot-starter-rocketmq

This project has been contributed to apache，please see：apache/rocketmq-externals/rocketmq-spring-boot-starter

中文

Help developers quickly integrate RocketMQ in Spring Boot. Support the Spring Message specification to facilitate developers to quickly switch from other MQ to RocketMQ.

Features:

• [x] synchronous transmission
• [x] synchronous ordered transmission
• [x] asynchronous transmission
• [x] asynchronous ordered transmission
• [x] orderly consume
• [x] concurrently consume(broadcasting/clustering)
• [x] One-way transmission
• [ ] transaction transmission
• [ ] Pull consume

Quick Start

<!--add dependency in pom.xml-->
<dependency>
    <group>com.qianmi</group>
    <artifactId>spring-boot-starter-rocketmq</artifactId>
    <version>1.1.0-RELEASE</version>
</dependency>

Produce Message

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

Note:

Maybe you need change 127.0.0.1:9876 with your real NameServer address for RocketMQ

@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;
    }
}

More relevant configurations for produce:

spring.rocketmq.producer.retry-times-when-send-async-failed=0
spring.rocketmq.producer.send-msg-timeout=300000
spring.rocketmq.producer.compress-msg-body-over-howmuch=4096
spring.rocketmq.producer.max-message-size=4194304
spring.rocketmq.producer.retry-another-broker-when-not-store-ok=false
spring.rocketmq.producer.retry-times-when-send-failed=2

Consume Message

## application.properties
spring.rocketmq.name-server=127.0.0.1:9876

Note:

Maybe you need change 127.0.0.1:9876 with your real NameServer address for RocketMQ

@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);
        }
    }
}

More relevant configurations for consume:

see: RocketMQMessageListener

FAQ

1. How to connected many nameserver on production environment？

   spring.rocketmq.name-server support the configuration of multiple nameserver, separated by ;. For example: 172.19.0.1: 9876; 172.19.0.2: 9876

2. When was rocketMQTemplate destroyed?

   Developers do not need to manually execute the rocketMQTemplate.destroy () method when using rocketMQTemplate to send a message in the project, and rocketMQTemplate will be destroyed automatically when the spring container is destroyed.

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

   RocketMQ in the design do not want a consumer to deal with multiple types of messages at the same time, so the same consumerGroup consumer responsibility should be the same, do not do different things (that is, consumption of multiple topics). Suggested consumerGroup and topic one correspondence.

4. How is the message content body being serialized and deserialized?

   RocketMQ's message body is stored as byte []. When the business system message content body if it is java.lang.String type, unified in accordance with utf-8 code into byte []; If the business system message content is not java.lang.String Type, then use jackson-databind serialized into the JSON format string, and then unified in accordance with utf-8 code into byte [] .

5. How do I specify the tags for topic?

   RocketMQ best practice recommended: an application as much as possible with one Topic, the message sub-type with tags to identify, tags can be set by the application free.

   When you use rocketMQTemplate to send a message, set the destination of the message by setting the destination parameter of the send method. The destination format is topicName:tagName, : Precedes the name of the topic, followed by the tags name.

   Note:

   tags looks a complex, but when sending a message , the destination can only specify one topic under a tag, can not specify multiple.

6. How do I set the message's key when sending a message?

   You can send a message by overloading method like xxxSend(String destination, Message<?> msg, ...), setting headers of msg. for example:

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

   Similarly, you can also set the message FLAG, WAIT_STORE_MSG_OK and some other user-defined other header information according to the above method.

   Note:

   In the case of converting Spring's Message to RocketMQ's Message, to prevent the header information from conflicting with RocketMQ's system properties, the prefix USERS_ was added in front of all header names. So if you want to get a custom message header when consuming, please pass through the key at the beginning of USERS_ in the header.

7. When consume message, in addition to get the message payload, but also want to get RocketMQ message of other system attributes, how to do?

   Consumers in the realization of RocketMQListener interface, only need to be generic for the MessageExt can, so in the onMessage method will receive RocketMQ native 'MessageExt` message.

   @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. How do I specify where consumers start consuming messages?

   The default consume offset please refer: RocketMQ FAQ. To customize the consumer's starting location, simply add a RocketMQPushConsumerLifecycleListener interface implementation to the consumer class. Examples are as follows:

   @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()));
       }
   }
   

   Similarly, any other configuration on DefaultMQPushConsumer can be done in the same way as above.