Spring 云数据流上的 PolledProcessor 问题

Question

我正在使用 PolledProcessor 实现一个 Spring Cloud 数据流处理器。我按照这里的例子https://spring.io/blog/2018/02/27/spring-cloud-stream-2-0-polled-consumers。下面是我的代码。我将一个带有源管道的流部署到这个处理器（源 | polled-processor）到 scdf，并让源发布了一些消息。我确认处理器每秒都会从 scdf rabbitmq 轮询消息，但result 始终是false。我去了 scdf rabbitmq 控制台，我看到那些消息都在队列中。因此，尽管处理器在代码中不断轮询，但它并没有轮询消息。我还看到队列没有消费者。看起来 scdf 没有将此处理器绑定到队列。知道为什么吗？

public interface PolledProcessor {
    @Input
    PollableMessageSource source();

    @Output
    MessageChannel dest();
}

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

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

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

这里是源和处理器之间的队列状态

Answer 1

我已经测试了一个没有问题的 Spring Cloud Stream 应用程序：

@SpringBootApplication
@EnableBinding(Polled.class)
public class So69383266Application {

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

    @Bean
    public ApplicationRunner runner(PollableMessageSource source) {
        return args -> {
            while (true) {
                boolean result = source.poll(System.out::println);
                System.out.println(result);
                Thread.sleep(1000);
            }
        };
    }

}

interface Polled {

    @Input
    PollableMessageSource source();

}

false
GenericMessage [payload=byte[6], headers={...
true
false

我建议你在AmqpMessageSource.doReceive() 中设置一个断点，看看发生了什么。

编辑

以下是检查源是否从正确的队列中消费的方法：

@Bean
public ApplicationRunner runner(PollableMessageSource source) {
    return args -> {
        while (true) {
            DirectFieldAccessor dfa = new DirectFieldAccessor(source);
            log.info(dfa.getPropertyValue("source.h.advised.targetSource.target.queue").toString());
            boolean result = source.poll(System.out::println);
            System.out.println(result);
            Thread.sleep(1000);
        }
    };
}