如何自定义ID生成@kafkalistener？

Question

我需要能够根据@kafkalistener的每个方法自定义ID，并基于其属性之一的值以及application.yaml中定义的值。 ，例如：

类似方法的方法的课程。

@KafkaListener(info="myInfo")
public void listen(String msg){

}

在我的应用程序中

myapp:
  myProperty: myProp

拥有一个具有 当我不明确提供注释属性时，生产的自动生成。

实现这一目标的最佳方法是什么？我正在考虑扩展kafkalistenerendpointregistrar或beanpostProcessor？

谢谢

Answer 1

在调用超级类之前，覆盖端点注册表bean并在那里操纵端点属性。

这是一个例子：

@SpringBootApplication
public class So72719215Application {

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

    @Bean(KafkaListenerConfigUtils.KAFKA_LISTENER_ENDPOINT_REGISTRY_BEAN_NAME)
    KafkaListenerEndpointRegistry registry(@Value("${myApp.myProperty}") String myProperty) {
        return new KafkaListenerEndpointRegistry() {

            @Override
            public void registerListenerContainer(KafkaListenerEndpoint endpoint,
                    KafkaListenerContainerFactory<?> factory) {

                AbstractKafkaListenerEndpoint<?, ?> akle = (AbstractKafkaListenerEndpoint<?, ?>) endpoint;
                akle.setId(new String(akle.getListenerInfo()) + "_" + myProperty);
                akle.setGroupId("group_" + myProperty);
                super.registerListenerContainer(endpoint, factory);
            }

        };
    }

    @KafkaListener(topics = "so72719215", info = "foo")
    void listen(String in) {
        System.out.println(in);
    }

    @Bean
    public NewTopic topic() {
        return TopicBuilder.name("so72719215").partitions(1).replicas(1).build();
    }

    @Bean
    ApplicationRunner runner(KafkaListenerEndpointRegistry reg) {
        return args -> {
            System.out.println(reg.getListenerContainerIds());
        };
    }

}

Answer 2

请参阅该id（） @kafkalistener的属性的Javadocs：

/**
 * The unique identifier of the container for this listener.
 * <p>If none is specified an auto-generated id is used.
 * <p>Note: When provided, this value will override the group id property
 * in the consumer factory configuration, unless {@link #idIsGroup()}
 * is set to false or {@link #groupId()} is provided.
 * <p>SpEL {@code #{...}} and property place holders {@code ${...}} are supported.
 * @return the {@code id} for the container managing for this endpoint.
 * @see org.springframework.kafka.config.KafkaListenerEndpointRegistry#getListenerContainer(String)
 */
String id() default "";

因此，您可以使用SPEL和属性占位符为目标侦听器容器定义动态ID。像：

@KafkaKListener(id = "#{myBean.generateId('${property.from.env}')}")