对在KafkaListener中使用并发属性创建的每个卡夫卡使用者线程进行预处理

Question

我第一次在一个spring引导应用程序中构建了一个kafka消费者，所以请用参考资料或示例代码片段来指导我。

我的要求如下：

应用程序应该有5-6个使用者，即线程->中的每个使用者一个，我可以使用@KafkaListener注释中的并发属性来设置它。

我必须在kafka使用者启动之前创建一些对象来处理消息，并且每个线程应该有一个新的对象，因为这些对象在db ->中执行不同的写操作--请建议我如何通过在@KafkaListener中设置并发属性来实现这一点。

如果我必须自己实例化线程，我会将这个对象初始化/预处理步骤放在Runnable接口的run方法中，但是当使用spring时，即当由于并发属性而初始化一个新线程时，会自动调用一个方法(类似于run方法的Runnable接口)来进行预处理。

Answer 1

您可以添加一个@EventListener方法来使用ConsumerStartedEvent的。

https://docs.spring.io/spring-kafka/docs/current/reference/html/#events

ConsumerStartingEvent -在用户线程第一次启动时，在它开始轮询之前发布。

https://docs.spring.io/spring-kafka/docs/current/reference/html/#event-consumption