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

测试：