SpringCloudStream整合RabbitMQ用ttl+死信实现延迟队列的实践

这篇是关于我使用Spring Cloud Steam操作RabbitMQ采用ttl+死信队列的方式实现的延迟队列。

前言

在公司项目中遇到了需要延迟队列的需求，为了以后可维护性和扩展性要求必须要用Springcloud Stream组件来操作mq，而且公司的rabbit也不允许安装延迟插件，只能用最原始的ttl+死信来实现，在搭建过程中遇到很多问题，最终成功实现，下面是代码，采用的是spring cloud steam3.1后的函数式编程实现。

先科普下原理：生产者发送消息到普通交换机绑定了个设置ttl时间的队列，这个队列绑定了个死信交换机且没人消费，如果消息过期就会发送到死信队列里，消费者就监听这个死信队列实现延迟队列的效果。

yaml配置文件

function中definition就是定义你的函数名称（后面发送和接收方法的名字）

bindings中的xxx-in/out-0，分为三个部分。第一个部分就是方法名，第二部分是输入或输出代表生产者out还是消费者in的意思，第三部分的0在rabbitmq里面是固定的，他是为了兼容kafkaf提供的，我们死写0就行。destination里面配置交换机的名字，后面的变量是环境的意思，到时候会变成dev、test、uat这些。group分组要设置一下我这里就叫模块的名字了。

**重点:**下面的producer：required-groups一定要配置，就是那个分组名字。

然后就是配置消费者，绑定的交换机要是最下面为生产者配置的死信交换机，分组名字也要记得填上不然不会消费。

后面的content-type是类型，可以不指定，默认就是这个

最下面是rabbit的配置，bingdings里面我配置了为生产者邦迪一个死信交换机，然后设置生产者多久没消费就会到死信的ttl。dead-letter-exchange是指定交换机名字，你就设置为原本交换机名称_DLX就好了为了规范，然后配置死信队列的名字，就是在交换机后面加上那个分组，一定要和上面的分组一致。

因为这里死信队列出来的消息是direct的，消费者是监听死信队列已经过期的消息的，所以交换机类型也要设置为direct，不然会报：direct消息转化topic异常，但是这个异常只会第一个消息报，后面的消息都不会了，可能是springcloud stream底层帮我们处理了，但为了避免第一次报错还是设置一下交换机类型。

  cloud:
    function:
      definition: memberAccountUpdateTaskDelayed;handleTaskDelayed;
    stream:
      bindings:
        memberAccountUpdateTaskDelayed-out-0:
          destination: MEMBER_TOPIC_${spring.profiles.active} #延迟exchange,交换模式是topic
          content-type: application/json #设置消息的类型为json
          group: ${spring.application.name}
          producer:
            required-groups: ${spring.application.name} #必须要制定生产者分组,不然发不过去
        handleTaskDelayed-in-0:
          destination: MEMBER_TOPIC_DLX_${spring.profiles.active}
          content-type: application/json
          group: ${spring.application.name}
      rabbit:
        bindings:
          memberAccountUpdateTaskDelayed-out-0:
            producer:
              ttl: 10000 #延时队列的延时时间，单位毫秒
              auto-bind-dlq: true #开启死信队列
              dead-letter-exchange: MEMBER_TOPIC_DLX_${spring.profiles.active} #死信交换机
              dead-letter-queueName: MEMBER_TOPIC_DLX_${spring.profiles.active}.${spring.application.name} #死信队列名称
          handleTaskDelayed-in-0:
            consumer:
              exchange-type: direct #死信交换机必须是direct类型的才能接受

生产者发送消息

这里我发送的是对象，在TaskDelayMessage就是发送的对象，实体类里面的属性上记得加上JSON反序列化的注解，不然消费者监听的时候会报反序列化错误。

方法名要和配置文件你配置的那个方法名一样

@Repository
@Slf4j
public class MemberAccountUpdateTaskHandleTimeoutTaskRepositoryImpl implements MemberAccountUpdateTaskHandleTimeoutTaskRepository {

    private final Sinks.Many<Message<TaskDelayMessage>> sinks =
            Sinks.many().multicast().onBackpressureBuffer();

    @Bean
    public Supplier<Flux<Message<TaskDelayMessage>>> memberAccountUpdateTaskDelayed(){
        return sinks::asFlux;
    }

    @Override
    public void sendDelayMessage(TaskDelayMessage message) {
        log.info("生产者准备发送消息:{}", message+" -"+System.currentTimeMillis());
        Message<TaskDelayMessage> msg = MessageBuilder.withPayload(message).build();
        while (sinks.tryEmitNext(msg).isFailure()) {
            LockSupport.parkNanos(10);
        }
        log.info("生产者成功发送消息:{}", message+" -"+System.currentTimeMillis());
    }
}

消费者监听消息

方法名要和配置文件你配置的那个方法名一样

    @Bean
    Consumer<TaskDelayMessage> handleTaskDelayed() {
        return message-> {
            log.info("消费者监听到了消息:{}", message);
            memberAccountUpdateTaskService.handleTimeOutTasks(message);
        };
    }

总结

总结：其实就是配置麻烦，国内springcloud steam的教程和博客是真的很少，搭建过程中遇到了很多问题csdn和b站也都没找到解决办法。最后看了油管和springcloud官方文档还有追进去看源码才解决。如果想学习的stream这个组件推荐去油管看教程，或者直接看springcloud官网文档。