暂停来自主 kafka 流的消息消费并从其他 kafka 主题开始

Question

我正在使用@StreamListener（spring cloud stream）来消费来自主题（input-channel）的消息，进行一些处理并保存到某个缓存或数据库中。

我的要求是，如果 DB 在处理消费的消息时出现故障，我想暂停主要消费者（输入通道），并从另一个 TOPIC（INPUT56-CHANNEL）开始消费，并且一旦它消耗所有来自 INPUT56-CHANNEL 的消息（没有很多），我想再次恢复主要消费者（输入通道）。

这能实现吗？？

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));
        }
    }

}