Spring Cloud Stream 定時任務消息延遲隊列

定時任務

每天定時執行任務可以使用springboot里面提供的@Schelduled注解加上cron表達式，但是有一些特殊的任務，比如幾分鍾之后執行，像這類定時任務可以使用Spring Cloud Stream+RabbitMQ來實現
這個首先要下載rabbitmq插件

啟動docker：service docker start
安裝rabbitmq：docker run -d --hostname my-rabbit --name yybb-rabbit -p 5672:5672 -p 15672:15672 rabbitmq:3-management
啟動rabbitmq：docker start 容器ID

1、把插件上傳至Linux,創建目錄rabbitmq_delayed_message
2、解壓：unzip rabbitmq_delayed_message_exchange-20171201-3.7.x.zip
3、將解壓文件copy至docker容器：docker cp /rabbitmq_delayed_message/rabbitmq_delayed_message_exchange-20171201-3.7.x.ez yybb-rabbit:/plugins
4、進入容器：docker exec -it yybb-rabbit /bin/bash
5、啟動插件：rabbitmq-plugins enable rabbitmq_delayed_message_exchange
6、查看插件是否啟動成功：rabbitmq-plugins list

創建boot工程，需要引入以下依賴

創建通道接口：

/**
 * @Author: Ben
 * @CreateTime: 2020-04-25 17:33
 */
public interface MyChannel {
    String INPUT = "yybb-input";
    String OUTPUT = "yybb-output";

    @Output(OUTPUT)
    MessageChannel output();

    @Input(INPUT)
    SubscribableChannel input();
}

創建消息接收器並綁定通道：

/**
 * @Author: Ben
 * @CreateTime: 2020-04-25 17:03
 * 消息接收器
 */
@EnableBinding(MyChannel.class) //通道綁定
public class MsgReceiver2 {

    public final static Logger logger = getLogger(MsgReceiver2.class);

    @StreamListener(MyChannel.INPUT)
    public void receiver(Object payload) {
        logger.info("receiverd2:"+payload+new Date());
    }
}

創建controller

 	//消息延遲6秒
    @GetMapping("/hello2")
    public void hello2(){
        logger.info("send msg :"+new Date());
        myChannel.output().send(MessageBuilder.withPayload("hello spring cloud stream !!")
                .setHeader("x-delay", 6000).build());
    }

編寫配置文件

spring.rabbitmq.host=192.168.238.128
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest

spring.cloud.stream.bindings.yybb-input.destination=topic
spring.cloud.stream.bindings.yybb-output.destination=topic
##
#spring.cloud.stream.bindings.yybb-input.destination=delay_msg
#spring.cloud.stream.bindings.yybb-output.destination=delay_msg

# 開啟消息延遲功能
spring.cloud.stream.rabbit.bindings.yybb-input.consumer.delayed-exchange=true
spring.cloud.stream.rabbit.bindings.yybb-output.producer.delayed-exchange=true

測試：