PolledProcessor在Spring云数据流上的问题

Question

我正在使用PolledProcessor提出一个spring云数据流处理器。我遵循了下面的示例https://spring.io/blog/2018/02/27/spring-cloud-stream-2-0-polled-consumers。下面是我的代码。我将带有源管道的流部署到这个处理器（源polled-processor）到scdf，并让源发布一些消息。我确认处理器每秒轮询一次来自scdf rabbitmq的消息，但是结果总是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);
        }
    };
}