@KafkaListener：通过传入参数生成clientId和groupId

如何解决@KafkaListener：通过传入参数生成clientId和groupId

我一直将@KafkaListeners 用于必须具有唯一 ID 的 Kafka 事件消费者，并且每个事件消费者必须有一个具有非常特殊名称的单独组。 例如：

@KafkaListener(id="AmazingProductEventConsumer",topics = "${kafka......topic}",clientIdPrefix = "AmazingProductEventConsumerClientId",groupId = "${kafka.group.id.prefix}-${environemnt.id}-${application.name}-${kafka.......topic}-#{T(java.util.UUID).randomUUID().toString()}")
public class AmazingProductEventConsumer {
   ... methods...
}

或使用批处理事件侦听器：

public class ProductBatchEventConsumer {

   @KafkaListener(id="ProductBatchEventConsumer ",clientIdPrefix = "ProductBatchEventConsumerClientId",groupId = "${kafka.group.id.prefix}-${environemnt.id}-${application.name}-${kafka.......topic}-#{T(java.util.UUID).randomUUID().toString()}")
   public void batchEventConsumer(List<Record> records) {
     .... 
   }
}

应用程序相当大，所以很多消费者订阅了同一个主题，但一般来说，在多个服务中大约有 500-600 个消费者订阅了大约 180-200 个主题。我想避免使用样板代码并提取可用于生成这些参数的通用模式，主题和分区除外。

在 Spring Boot 2.2 中，我为此使用了一个单独的 BeanPostProcessor 并在初始化之前生成了所需的字段，并用需要的实例替换了消费者的 KafkaListener 注释，但在 Spring Boot 2.5.0 中，此功能不再可用，因为 KafkaListenerAnnotationBeanPostProcessor uses KafkaListener ann = AnnotatedElementUtils.findMergedAnnotation(clazz,KafkaListener.class); 基于首次声明的注释属性合成新的 KafkaListeners。

我很好奇，关于如何生成@KafkaListener 的字段id、clientId、groupId 和其他字段，是否有任何合法的方法可以通过通用模式从其他侦听器中提取出来。

解决方法

您可以使用更简单的 SpEL。

groupId = "#{@someBean.groupId}"

并将占位符和 UUID 生成放在该 bean 的 getGroupId() 中。

并且您可以使用带有 @KafkaListener 元注释的自定义注释和您的道具以避免样板。

或者，如果您对每个侦听器（UUID 除外）有其他一些可变性，请使用

#{@someBean.groupId('${small.property}'})