Spring Cloud Stream 使用延迟消息实现定时任务（RabbitMQ）

ゞ浴缸里的玫瑰 2024-04-18 19:43 199阅读 0赞

开发项目中，有的服务对于任务的执行时间通常都是有规律性的，可能是每隔半小时执行一次，或者每天凌晨一点执行一次。这些我们通常都是使用**[*SpringQuartz* 、springboot下使用][SpringQuartz_ _springboot]Schedule**然而实际业务中还存在另外一种定时任务，它可能需要一些触发条件才开始定时，例如 编写文本时候定时几分钟保存一次的草稿。过期取消订单

对于这些开始时间不确定的定时任务，可以通过Spring Cloud Stream来很好的处理。延迟消息使用RabbitMQ中提供了关于延迟消息的插件。

安装RabbitMQ

[https://www.rabbitmq.com/download.html][https_www.rabbitmq.com_download.html]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h1ZXhpamF2YTg1_size_16_color_FFFFFF_t_70][]

rabbitMQ是一个在AMQP协议标准基础上完整的，可服用的企业消息系统。它遵循Mozilla Public License开源协议，采用 Erlang 实现的工业级的消息队列(MQ)服务器，Rabbit MQ 是建立在Erlang OTP平台上。

需要先安装erlang 安装过程会提示下载 ![2019091111221118.png][]进行安装

在安装好RabbitMQ 进行环境变量配置，新建RABBITMQ\_SERVER  路径 D:\\Program Files\\RabbitMQ3.7.8\\rabbitmq\_server-3.7.8\\sbin

**可视化界面启动**

1.  运行命令行窗口cmd
2.  输入命令rabbitmq-plugins enable rabbitmq\_management，这样就可以添加可视化插件了。
3.  ![20190911112624613.png][]

只要启动rabbitmq，然后在浏览器输入 http://127.0.0.1:15672/ 就可以访问了

用户名密码 guest

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h1ZXhpamF2YTg1_size_16_color_FFFFFF_t_70 1][]

[http://www.rabbitmq.com/community-plugins.html][http_www.rabbitmq.com_community-plugins.html] 中找到`rabbitmq_delayed_message_exchange`插件   只有RabbitMQ 3.6.x以上才支持

在下载好之后，解压得到`.ez`结尾的插件包，将其复制到RabbitMQ安装目录下的`plugins`文件夹。

然后通过命令行启用该插件：

rabbitmq-plugins enable rabbitmq_delayed_message_exchange

pom

<parent>  
        <groupId>org.springframework.boot</groupId>  
        <artifactId>spring-boot-starter-parent</artifactId>  
        <version>2.0.5.RELEASE</version>  
        <relativePath/>   
    </parent>

<dependencies>  
  <dependency>  
            <groupId>org.projectlombok</groupId>  
            <artifactId>lombok</artifactId>  
            <version>1.18.2</version>  
        </dependency>  
        <dependency>  
            <groupId>org.springframework.boot</groupId>  
            <artifactId>spring-boot-starter-test</artifactId>  
            <scope>test</scope>  
        </dependency>

<dependency>  
            <groupId>org.springframework.cloud</groupId>  
            <artifactId>spring-cloud-starter-stream-rabbit</artifactId>  
        </dependency>  
        <dependency>  
            <groupId>org.springframework.boot</groupId>  
            <artifactId>spring-boot-starter-actuator</artifactId>  
        </dependency>  
    <dependency>  
      <groupId>junit</groupId>  
      <artifactId>junit</artifactId>  
      <version>3.8.1</version>  
      <scope>test</scope>  
    </dependency>  
  </dependencies>  
  <dependencyManagement>  
        <dependencies>  
            <dependency>  
                <groupId>org.springframework.cloud</groupId>  
                <artifactId>spring-cloud-dependencies</artifactId>  
                <version>Finchley.SR1</version>  
                <type>pom</type>  
                <scope>import</scope>  
            </dependency>  
        </dependencies>  
    </dependencyManagement>

配置文件

spring.application.name=stream-delayed-message  
server.port=18081

spring.cloud.stream.bindings.example-topic-input.destination = delay-topic  
spring.cloud.stream.bindings.example-topic-input.group=test  
spring.cloud.stream.rabbit.bindings.example-topic-input.consumer.delayed-exchange=true

spring.cloud.stream.bindings.example-topic-output.destination=delay-topic  
spring.cloud.stream.rabbit.bindings.example-topic-output.producer.delayed-exchange=true

用来开启延迟消息的功能，这样在创建exchange的时候，会将其设置为具有延迟特性的exchange，也就是用到上面我们安装的延迟消息插件的功能

interface Topic \{  
        String OUTPUT = "example-topic-output";  
        String INPUT = "example-topic-input";

@Output(OUTPUT)  
        MessageChannel output();

@Input(INPUT)  
        SubscribableChannel input();

\}  
      
    @Slf4j  
    @RestController  
    public class Controller \{

@Autowired  
        private Topic topic;

/\*\*  
         \* 消息生产接口  
         \*/  
        @GetMapping("/sendMessage")  
        public String messageWithMQ(@RequestParam String message) \{  
            log.info("Send: "+message);  
            topic.output().send(MessageBuilder.withPayload(message).setHeader("x-delay",5000).build());  
            return "ok";  
        \}

\}  
      
     /\*\*  
     \* 消息消费逻辑  
     \*/  
    @Slf4j  
    @Component  
    publiclass TestListener \{

@StreamListener(Topic.INPUT)  
        public void receive(String payload) \{  
            log.info("Received: " + payload);  
        \}

跑起来 [http://localhost:18081/sendMessage?message=hello][http_localhost_18081_sendMessage_message_hello]

![20190911114018549.png][]

[SpringQuartz_ _springboot]: http://www.baidu.com/link?url=57aywD0Q6WTnl7XKbIHuE7lcWGXh50Vy3z1lItKlmd4OJre3dLQnWbn8s4nBPmV1JvNpRiZS8wxZ15gFG5xCFVgBRQRWgfj2FXrfrrgvxNa
[https_www.rabbitmq.com_download.html]: https://www.rabbitmq.com/download.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h1ZXhpamF2YTg1_size_16_color_FFFFFF_t_70]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/eaf5c22d13ce4dada2565e17ea4b4d00.png
[2019091111221118.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/95cf24b1e97544379ca0a893dfa46a85.png
[20190911112624613.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/4d7e79e9db3e4c17a6280f8bc2d6ddaa.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h1ZXhpamF2YTg1_size_16_color_FFFFFF_t_70 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/f8fccba0c38e40ebac9b1de728f8149e.png
[http_www.rabbitmq.com_community-plugins.html]: http://www.rabbitmq.com/community-plugins.html
[http_localhost_18081_sendMessage_message_hello]: http://localhost:18081/sendMessage?message=hello
[20190911114018549.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/18/5f7eec685de2491ab026c41dc66d35b9.png