1. Prerequisites in Alibaba Cloud

Before integration, provision a commercial RocketMQ instance and create the flolowing resources:

• Standard topic and consumer group for plain messages.
• Dedicated delay topic and cosnumer group for scheduled/delay messages.

2. Project Structure

The example follows a typical Spring Boot project layout; configuration classes are placed under config, listeners under listener, and properties under config.

3. Dependency

Add the Alibaba Cloud ONS client to your pom.xml:

<dependency>
    <groupId>com.aliyun.openservices</groupId>
    <artifactId>ons-client</artifactId>
    <version>1.8.8.5.Final</version>
</dependency>

4. Delay Messsage Consumer Setup

4.1 Batch Consumer Bean

import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.batch.BatchMessageListener;
import com.aliyun.openservices.ons.api.bean.BatchConsumerBean;
import com.aliyun.openservices.ons.api.bean.Subscription;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

@Configuration
public class DelayConsumerConfiguration {

    @Autowired
    private RocketMqProperties mqProperties;

    @Autowired
    private DelayBatchListener delayBatchListener;

    @Bean(initMethod = "start", destroyMethod = "shutdown")
    public BatchConsumerBean delayBatchConsumer() {
        BatchConsumerBean consumerBean = new BatchConsumerBean();
        Properties props = mqProperties.buildBaseProperties();
        props.setProperty(PropertyKeyConst.GROUP_ID, mqProperties.getDelayGroupId());
        props.setProperty(PropertyKeyConst.ConsumeThreadNums, "20");
        consumerBean.setProperties(props);

        Map<Subscription, BatchMessageListener> subs = new HashMap<>();
        Subscription sub = new Subscription();
        sub.setTopic(mqProperties.getDelayTopic());
        sub.setExpression(mqProperties.getDelayTag());
        subs.put(sub, delayBatchListener);
        consumerBean.setSubscriptionTable(subs);
        return consumerBean;
    }
}

4.2 Delay Message Listener

import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.ConsumeContext;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.batch.BatchMessageListener;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.List;

@Slf4j
@Component
public class DelayBatchListener implements BatchMessageListener {

    private static final DateTimeFormatter FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

    @Override
    public Action consume(List<Message> msgs, ConsumeContext ctx) {
        log.info("Batch received {} messages", msgs.size());
        for (Message msg : msgs) {
            String payload = new String(msg.getBody(), StandardCharsets.UTF_8);
            String now = LocalDateTime.now().format(FORMATTER);
            System.out.println("Consume time: " + now);
            log.info("Payload: {}", payload);
        }
        try {
            // business processing
            return Action.CommitMessage;
        } catch (Exception ex) {
            log.error("Batch processing failed", ex);
            return Action.ReconsumeLater;
        }
    }
}

5. Centralised RocketMQ Properties

import com.aliyun.openservices.ons.api.PropertyKeyConst;
import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;

import java.util.Properties;

@Data
@Component
@ConfigurationProperties(prefix = "rocketmq")
public class RocketMqProperties {
    private String accessKey;
    private String secretKey;
    private String nameSrvAddr;
    private String topic;
    private String groupId;
    private String tag;
    private String orderTopic;
    private String orderGroupId;
    private String orderTag;
    private String delayTopic;
    private String delayGroupId;
    private String delayTag;

    public Properties buildBaseProperties() {
        Properties props = new Properties();
        props.setProperty(PropertyKeyConst.AccessKey, this.accessKey);
        props.setProperty(PropertyKeyConst.SecretKey, this.secretKey);
        props.setProperty(PropertyKeyConst.NAMESRV_ADDR, this.nameSrvAddr);
        return props;
    }
}