Spring Boot整合RocketMQ实现延迟消息消费

导包

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

添加配置信息

application配置文件

# rocketMq地址
rocketmq.name-server=106.52.60.215:9876
# 生产者分组
rocketmq.producer.group=myGroup
rocketmq.producer.topics=topic1
# 消费者分组
rocketmq.consumer.group=myGroup
# topic
rocketmq.consumer.topics=topic1
# 表示顺序消费模式
rocketmq.consumer.consume-mode=ORDERLY
# 消费者的最大线程数，即消费消息的线程池大小。默认值为20，如果不需要处理大量的消息，可以将其调小。
rocketmq.consumer.consume-thread-max=1
# 表示每次消费消息的最大数量，即一次性消费的最大消息数。默认值为1，即每次只消费一条消息。如果需要批量消费消息，可以将其调大。但是需要注意的是，批量消费消息可能会影响消费的效率和消息的顺序性。
rocketmq.consumer.consume-message-batch-max-size=1

yml配置文件

rocketmq:
  consumer:
    consume-message-batch-max-size: 1
    consume-mode: ORDERLY
    consume-thread-max: 1
    group: myGroup
    topics: topic1
  name-server: 106.52.60.215:9876
  producer:
    group: myGroup
    topics: topic1

生产者发送消息

同步发现消息

在Spring Boot中，可以使用RocketMQTemplate来发送消息。设置消息的延迟级别，可以使用RocketMQTemplate的send(Message message, long timeout, int delayLevel)方法，其中delayLevel为延迟级别，单位为秒

RocketMQ支持18个级别的延迟时间，分别为1s、5s、10s、30s、1m、2m、3m、4m、5m、6m、7m、8m、9m、10m、20m、30m、1h、2h

import com.songzixain.Blog;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;

/**
 * <p>
 * Description: 消息生产者
 * </p>
 *
 * @author 
 * @version 
 * @create 
 * @see 
 */
@Slf4j
@RestController
public class MyProducer1 {

    @Value("${rocketmq.producer.topics}")
    private String topic;

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    /**
     * description 同步发送延迟消息
     *
     * @param:  []
     * @return
     * @Date   2023/3/11
     */
    @GetMapping("syncSendTest")
    public void sendDelayMsg() {
        Blog blog = new Blog();
        blog.setBlogName("余十步");
        blog.setUrl("yushibu");

        // delayTimeLevel代表延迟级别  messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
        int delayTimeLevel = 3;
        Message<Blog> message = MessageBuilder.withPayload(blog)
                .setHeader(MessageConst.PROPERTY_DELAY_TIME_LEVEL, delayTimeLevel)
                .build();

        SendResult sendResult = rocketMQTemplate.syncSend(topic, message, 3000, delayTimeLevel);
        log.info("消息发送成功,时间:{} 发送内容:{}", LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")), blog.toString());
        log.info("发送结果:{}", sendResult);
    }
}

异步发送消息(推荐)

import com.songzixain.Blog;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;

/**
 * <p>
 * Description: 消息生产者
 * </p>
 *
 * @author 
 * @version 
 * @create 
 * @see 
 */
@Slf4j
@RestController
public class DemoProducers {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    /**
     * description 延迟消息发送
     *
     * @param:  [user]
     * @return
     * @Date   2023/3/11
    */
    @RequestMapping("/asyncSendTest")
     public  String asyncSendTest(){
        Blog blog = new Blog();
        blog.setBlogName("余十步");
        blog.setUrl("yushibu");
        // 构建消息体
        Message<Blog> msg = MessageBuilder.withPayload(blog).build();
        rocketMQTemplate.asyncSend("topic1", msg, new SendCallback() {
            // 发送成功
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("消息发送成功,时间:{} 发送内容:{}",  LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")), blog.toString());
            }
            // 发送失败
            @Override
            public void onException(Throwable throwable) {
                log.info("消息发送失败,时间:{} 发送内容:{}",  LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")), blog.toString());
            }
            // ps:3 代表第三个延迟10s   延迟级别:"1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h"
        },3000,3);

        return "发送成功";
    }
}

消息消费者

import com.songzixain.Blog;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;

/**
 * <p>
 * Description: 消息生产者
 * </p>
 *
 * @author 
 * @version 
 * @create 
 * @see 
@Slf4j
@RestController
public class MyProducer2 {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @Value("${rocketmq.producer.topics}")
    private String topic;

    /**
     * description 延迟消息发送
     * 在上面的代码中，我们使用了RocketMQTemplate的syncSend方法来发送消息。
     * 其中，第一个参数是消息的主题，第二个参数是消息内容，第三个参数是延迟时间（单位为毫秒）
     * ，第四个参数是发送消息的重试次数。
     * @param:  [user]
     * @return
     * @Date   2023/3/11
    */
    @RequestMapping("/asyncSendTest")
     public  String asyncSendTest(){
        Blog blog = new Blog();
        blog.setBlogName("余十步");
        blog.setUrl("yushibu");
        // 构建消息体
        Message<Blog> msg = MessageBuilder.withPayload(blog).build();
        rocketMQTemplate.asyncSend(topic, msg, new SendCallback() {
            // 发送成功
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("消息发送成功,时间:{} 发送内容:{}",  LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")), blog.toString());
            }
            // 发送失败
            @Override
            public void onException(Throwable throwable) {
                log.info("消息发送失败,时间:{} 发送内容:{}",  LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")), blog.toString());
            }
            // ps:3 代表第三个延迟10s   延迟级别:"messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h"
        },3000,4);

        return "发送成功";
    }
}