@StreamListener的自动启动

Question

与@KafkaListener不同，它看起来@StreamListener不支持autoStartup参数。有没有办法为@StreamListener实现相同的行为？这是我的用例：

我有一个通用的Spring应用程序，它可以监听任何Kafka主题并写入我数据库中的相应表。对于某些主题，音量较低，因此可以以非常低的延迟处理单个消息。对于其他大容量的主题，代码应该接收消息的微批处理并使用Jdbc批处理写入数据库的频率较低。理想情况下，侦听器的定义如下所示：

// low volume listener
@StreamListener(target = Sink.INPUT, autoStartup="${application.singleMessageListenerEnabled}")
public void handleSingleMessage(@Payload GenericRecord message) ...

// high volume listener
@StreamListener(target = Sink.INPUT, autoStartup="${application.multipleMessageListenerEnabled}")
public void handleMultipleMessages(@Payload List<GenericRecord> messageList) ...

对于小容量的主题，我会设置应用程序。SingleMessageListener已将启用为true和应用程序。multipleMessageListenerEnabled为false，对于大量主题，反之亦然。因此，只有一个听者会积极地听消息，而另一个不会积极地听。

有没有办法通过@StreamListener实现这一点？

Answer 1

首先，请考虑升级到函数式编程模型，这将花费您几分钟的时间进行重构。我们几乎都不赞成基于注释的编程模型。如果你这么做了，那么你想要完成的事情就很容易了：

@SpringBootApplication
public class SimpleStreamApplication {

    public static void main(String[] args) throws Exception {
        SpringApplication.run(SimpleStreamApplication.class);
    }

    @Bean
    public Consumer<GenericRecord> singleRecordConsumer() {...}

    @Bean
    public Consumer<List<GenericRecord>> multipleRecordConsumer() {...}
}

然后你可以简单地使用--spring。云作用定义=singleRecordConsumer单个案例的属性和--spring。云作用definition=multipleRecordConsumer启动应用程序时，这将确保要激活哪个特定的侦听器。