spring-boot-starter-rocketmq

中文

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:

• synchronous transmission
• synchronous ordered transmission
• asynchronous transmission
• asynchronous ordered transmission
• orderly consume
• concurrently consume(broadcasting/clustering)
• one-way transmission
• transaction transmission
• Pull consume

Quick Start

Here are some key points listed, the complete example, please refer to: rocketmq-demo

<!--add dependency in pom.xml-->
<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>spring-boot-starter-rocketmq</artifactId>
    <version>1.0.0</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
spring.rocketmq.topic=topic
spring.rocketmq.consumerGroup=group01
spring.rocketmq.consumeTime=2019-06-20 10:10:00

Note:

Maybe you need change 127.0.0.1:9876 with your real NameServer address for RocketMQ
If you want consume message starting at specified time ,Plase must set consumeTime for @RocketMQMessageListener. If you do not set it,
Consumer will consume message from the offset of the last consumption of this consumer group.

@SpringBootApplication
public class ConsumerApplication{
    public static void main(String[] args){
        SpringApplication.run(ConsumerApplication.class, args);
    }
    @Slf4j
    @Service
    @RocketMQMessageListener(topic = "${spring.rocketmq.topic}", consumerGroup = "${spring.rocketmq.consumerGroup}",consumeTime = "${spring.rocketmq.consumeTime}")
    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" consumeTime = "2019-06-20 10:10:00")
    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, androcketMQTemplate 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 andtopic 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 withutf-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 withutf-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 thedestination 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 theMessageExt 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.