springCloud使用stream配置rabbitMq实现延时消息

参考我另一篇文章
https://blog.csdn.net/weixin_43944305/article/details/120828003

   org.springframework.boot
   spring-boot-starter-web



   org.springframework.cloud
   spring-cloud-starter-stream-rabbit

消息通道

package com.fchan.springcloudstream.service;

import org.springframework.cloud.stream.annotation.Input;
import org.springframework.cloud.stream.annotation.Output;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.SubscribableChannel;

public interface MyMessageChannel {

    String out = "out";
    String in = "in";
    @Output(out)
    MessageChannel out();
    @Input(in)
    SubscribableChannel in();
}

 // 发送延迟消息
@PostMapping("/delayed")
public String sendDelayedMessage(@RequestParam("body") String body,
                                 @RequestParam("seconds") Integer seconds) {

    Map message = new HashMap<>();
    message.put("body", body);
    myMessageChannel.out().send(
            MessageBuilder.withPayload(message)
                    .setHeader("x-delay", seconds * 1000)
                    .build()
    );
    log.info("发送延迟消息成功");
    return "SUCCESS";
}

延时消息接收

package com.fchan.springcloudstream.service;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;

import java.util.Map;

@Component
@EnableBinding({MyMessageChannel.class})
public class MyConsumer {
    Logger log = LoggerFactory.getLogger(MyConsumer.class);
    @StreamListener(MyMessageChannel.in)
    public void input(Message> message){
        log.info("收到消息：{}", message.getPayload());
    }
}

yml配置

spring:
  rabbitmq:
    host: 110.40.181.73
    port: 35672
    username: root
    password: 10086
    virtual-host: /fchan
  cloud:
    stream:
      rabbit:
        bindings:
#          消费者开启延时队列支持
          in:
            consumer:
              delayed-exchange: true
#          生产者开启延时队列支持
          out:
            producer:
              delayed-exchange: true
      bindings:
        in:
#          指定消息所属exchange
          destination: test
#          指定消费者分组，在多实例的时候必需指定，防止重复消费
          group: myIn
        out:
          destination: test

在启动项目后登陆rabbitmq管理页面可以看到exchange创建成功

localhost:8080/delayed?body=1231&seconds=5
测试成功