从主Kafka流暂停消息消费，并从其他Kafka主题开始

Question

我正在使用@StreamListener（Spring-Cloud-Stream）来使用来自主题（输入通道）的消息，进行一些处理并保存到一些缓存或数据库中。

我的要求是，如果DB在处理消费的消息时停止，我想暂停主消费者(输入通道)，并从另一个主题(输入56通道)开始消费，一旦它消费了来自输入56通道的所有消息(没有很多)，我想再次恢复主消费者(输入通道)。

这能做到吗？？

Answer 1

不推荐使用@StreamListener；您应该转换为函数式编程模型。

下面是一个使用该模型的示例（但同样的技术也适用于不推荐使用的侦听器）。

spring.cloud.function.definition=input1;input2

spring.cloud.stream.bindings.input1-in-0.group=grp1
spring.cloud.stream.bindings.input2-in-0.consumer.auto-startup=false
spring.cloud.stream.bindings.input2-in-0.group=grp2

spring.cloud.stream.kafka.bindings.input2-in-0.consumer.idle-event-interval=5000

@SpringBootApplication
public class So69726610Application {

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

    boolean dbIsDown = true;

    @Autowired
    BindingsLifecycleController controller;

    TaskExecutor exec = new SimpleAsyncTaskExecutor();

    @Bean
    public Consumer<String> input1() {
        return str -> {
            System.out.println(str);
            if (this.dbIsDown) {
                this.controller.changeState("input1-in-0", State.PAUSED);
                this.controller.changeState("input2-in-0", State.STARTED);
                throw new RuntimeException("Paused");
            }
        };
    }

    @Bean
    public Consumer<String> input2() {
        return System.out::println;
    }

    @EventListener
    public void idle(ListenerContainerIdleEvent event) {
        System.out.println(event);
        // assumes concurrency = 1 (default)
        if (event.getListenerId().contains("input2-in-0")) {
            this.controller.changeState("input1-in-0", State.RESUMED);
            this.exec.execute(() -> this.controller.changeState("input2-in-0", State.STOPPED));
        }
    }

}