Spring Cloud Data Flow中PolledProcessor无法绑定队列消费消息问题咨询

根因分析 #

该问题的核心是自定义绑定通道的名称和SCDF标准处理器的默认绑定命名规则不匹配，导致SCDF没有将上游source的队列和自定义的输入通道做绑定。
具体细节：

1. Spring Cloud Stream的@Input/@Output注解如果不指定显式名称，默认会用方法名作为绑定的通道名称。当前PolledProcessor接口输入方法名为source()，输出方法名为dest()，因此对应的输入绑定名为source，输出绑定名为dest。
2. SCDF对于标准处理器类型的应用，默认要求输入绑定名为input、输出绑定名为output，部署时只会自动为这两个名称的绑定配置对应的队列和映射关系。自定义通道名不符合该规则，因此没有和上游队列绑定，队列无消费者，poll()方法自然始终返回false。

解决方案 #

可选择以下任意一种方式修复：

• 方案一：直接修改绑定接口的命名，对齐SCDF默认规则（最简便）
  修改PolledProcessor接口和注入逻辑：

public interface PolledProcessor {
    @Input
    PollableMessageSource input();

    @Output
    MessageChannel output();
}

@SpringBootApplication
@EnableBinding(PolledProcessor.class)
public class Application {

    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }

    @Bean
    public ApplicationRunner runner(PollableMessageSource input, MessageChannel output) {
        return args -> {
            while (true) {
                boolean result = input.poll(output::send);
                Thread.sleep(1000);
            }
        };
    }
}

• 方案二：手动配置绑定映射，适配现有接口
  如果不想修改现有接口代码，可以在应用的配置文件（application.properties/application.yml）中添加如下配置，将自定义绑定名和SCDF默认的绑定配置做映射：

# 输入通道绑定配置
spring.cloud.stream.bindings.source.destination=${spring.cloud.stream.bindings.input.destination}
spring.cloud.stream.bindings.source.group=${spring.cloud.stream.bindings.input.group}
# 输出通道绑定配置
spring.cloud.stream.bindings.dest.destination=${spring.cloud.stream.bindings.output.destination}

注意事项 #

• 当前参考的是Spring Cloud Stream 2.0版本的示例，目前Spring Cloud Stream 3.x及以上版本已经将@EnableBinding标记为废弃，推荐使用函数式编程模型实现轮询消费者，兼容性更好。
• 修复部署后可以再查看RabbitMQ控制台，确认对应队列已经绑定消费者，再验证消息消费逻辑即可。

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