Spring Cloud Stream队列过滤：新Consumers模式下如何实现消息条件过滤

原有@StreamListener的condition属性是基于SpEL表达式实现消息过滤，在3.x+版本推荐的函数式消费模型中，可以通过以下两种方式实现同等效果：

方案1：配置式过滤（无代码侵入，和原有用法对齐） #

该方案不需要修改业务消费代码，直接通过配置文件添加过滤规则，SpEL语法和原有condition完全兼容，只有符合规则的消息才会进入消费函数。

消费函数示例 #

import java.util.function.Consumer;
import org.springframework.context.annotation.Bean;
import org.springframework.messaging.Message;

public class CustomMessageConsumer {

    @Bean
    public Consumer<Message<String>> yourConsumer() {
        return message -> {
            // 只有符合过滤规则的消息才会执行到此处业务逻辑
            System.out.println("收到符合条件的消息：" + message.getPayload());
        };
    }
}

配置文件示例（application.yml） #

spring:
  cloud:
    stream:
      # 声明注册的消费函数，多函数用;分隔
      function:
        definition: yourConsumer
      bindings:
        # binding命名规则：消费函数名 + -in + 输入序号（单输入默认是0）
        yourConsumer-in-0:
          destination: 你的消息topic名称
          group: 你的消费组名称
      # 给指定binding配置过滤规则，SpEL语法和旧版condition完全一致
      filter:
        yourConsumer-in-0:
          expression: headers['header_name'] == 'header_value'

方案2：代码式过滤（灵活支持复杂逻辑） #

该方案直接在消费函数内部判断消息头，适合需要动态计算、逻辑复杂的过滤场景。

代码示例 #

import java.util.function.Consumer;
import org.springframework.context.annotation.Bean;
import org.springframework.messaging.Message;

public class CustomMessageConsumer {

    @Bean
    public Consumer<Message<String>> yourConsumer() {
        return message -> {
            // 手动实现过滤逻辑，不符合要求的消息直接跳过处理
            Object headerValue = message.getHeaders().get("header_name");
            if (!"header_value".equals(headerValue)) {
                return;
            }
            // 符合条件的消息执行业务逻辑
            System.out.println("收到符合条件的消息：" + message.getPayload());
        };
    }
}

方案选型建议 #

• 过滤规则简单、需要可配置的场景优先选配置式过滤，和旧用法兼容度最高
• 过滤逻辑复杂、需要依赖其他业务Bean判断的场景选代码式过滤，灵活性更强

内容的提问来源于stack exchange，提问作者Marx