代码之家  ›  专栏  ›  技术社区  ›  Barath

spring kafka:使用ReplyingKafkatTemplate的自动配置问题

spring-kafka
  •  1
  • Barath  · 技术社区  · 7 年前

    问题 :作为制片人,如果我想使用 , 我只需要配置 但自动配置失败,日志如下

    kafkanconfiguration.java : 

            @Autowired
        private KafkaProperties properties;

        @Bean
        public ConsumerFactory<Long, String> cf() {

          return new DefaultKafkaConsumerFactory<> 
             (properties.getConsumer().buildProperties());
        }

        @Bean
        public KafkaMessageListenerContainer<String, String> 
          container() {

            ContainerProperties props = new ContainerProperties("replies");
            KafkaMessageListenerContainer<Long, String> container
              = new KafkaMessageListenerContainer<>(cf(),props);             
            return container;
        }



        @Bean
        public ReplyingKafkaTemplate<Long, String, String> replyingKafkaTemplate(KafkaMessageListenerContainer<Long, String> container ){               
            return new ReplyingKafkaTemplate<Long, Order, String>(pf()), container);
        }

    卡夫卡说明驱动配置 导致此问题的原因: 

        @Bean
    @ConditionalOnMissingBean(name = "kafkaListenerContainerFactory")
    public ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory(
            ConcurrentKafkaListenerContainerFactoryConfigurer configurer,ConsumerFactory<Object, Object> kafkaConsumerFactory) {}


    org.springframework.beans.factory.UnsatisfiedDependencyException: Error creating bean with name 'kafkaListenerContainerFactory' defined in class path resource [org/springframework/boot/autoconfigure/kafka/KafkaAnnotationDrivenConfiguration.class]: Unsatisfied dependency expressed through method 'kafkaListenerContainerFactory' parameter 1; nested exception is org.springframework.beans.factory.NoSuchBeanDefinitionException: No qualifying bean of type 'org.springframework.kafka.core.ConsumerFactory<java.lang.Object, java.lang.Object>' available: expected at least 1 bean which qualifies as autowire candidate. Dependency annotations: {}
    at org.springframework.beans.factory.support.ConstructorResolver.createArgumentArray(ConstructorResolver.java:767) ~[spring-beans-5.1.2.RELEASE.jar:5.1.2.RELEASE]
    at org.springframework.beans.factory.support.ConstructorResolver.instantiateUsingFactoryMethod(ConstructorResolver.java:508) ~[spring-beans-5.1.2.RELEASE.jar:5.1.2.RELEASE]
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.instantiateUsingFactoryMethod(AbstractAutowireCapableBeanFactory.java:1288) ~[spring-beans-5.1.2.RELEASE.jar:5.1.2.RELEASE]
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.createBeanInstance(AbstractAutowireCapableBeanFactory.java:1127) ~[spring-beans-5.1.2.RELEASE.jar:5.1.2.RELEASE]
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.doCreateBean(AbstractAutowireCapableBeanFactory.java:538) ~[spring-beans-5.1.2.RELEASE.jar:5.1.2.RELEASE]
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.createBean(AbstractAutowireCapableBeanFactory.java:498) ~[spring-beans-5.1.2.RELEASE.jar:5.1.2.RELEASE]
    at org.springframework.beans.factory.support.AbstractBeanFactory.lambda$doGetBean$0(AbstractBeanFactory.java:320) ~[spring-beans-5.1.2.RELEASE.jar:5.1.2.RELEASE]
    at org.springframework.beans.factory.support.DefaultSingletonBeanRegistry.getSingleton(DefaultSingletonBeanRegistry.java:222) ~[spring-beans-5.1.2.RELEASE.jar:5.1.2.RELEASE]
    at org.springframework.beans.factory.support.AbstractBeanFactory.doGetBean(AbstractBeanFactory.java:318) ~[spring-beans-5.1.2.RELEASE.jar:5.1.2.RELEASE]
    at org.springframework.beans.factory.support.AbstractBeanFactory.getBean(AbstractBeanFactory.java:199) ~[spring-beans-5.1.2.RELEASE.jar:5.1.2.RELEASE]
    at org.springframework.beans.factory.support.DefaultListableBeanFactory.preInstantiateSingletons(DefaultListableBeanFactory.java:846) ~[spring-beans-5.1.2.RELEASE.jar:5.1.2.RELEASE]
    at org.springframework.context.support.AbstractApplicationContext.finishBeanFactoryInitialization(AbstractApplicationContext.java:863) ~[spring-context-5.1.2.RELEASE.jar:5.1.2.RELEASE]
    at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:546) ~[spring-context-5.1.2.RELEASE.jar:5.1.2.RELEASE]
    at org.springframework.boot.web.servlet.context.ServletWebServerApplicationContext.refresh(ServletWebServerApplicationContext.java:140) ~[spring-boot-2.1.0.RELEASE.jar:2.1.0.RELEASE]
    at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:775) [spring-boot-2.1.0.RELEASE.jar:2.1.0.RELEASE]
    at org.springframework.boot.SpringApplication.refreshContext(SpringApplication.java:397) [spring-boot-2.1.0.RELEASE.jar:2.1.0.RELEASE]
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:316) [spring-boot-2.1.0.RELEASE.jar:2.1.0.RELEASE]
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:1260) [spring-boot-2.1.0.RELEASE.jar:2.1.0.RELEASE]
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:1248) [spring-boot-2.1.0.RELEASE.jar:2.1.0.RELEASE]
    at com.barath.app.KafkaProducerApplication.main(KafkaProducerApplication.java:10) [classes/:na]
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:1.8.0_101]
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[na:1.8.0_101]
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[na:1.8.0_101]
    at java.lang.reflect.Method.invoke(Method.java:498) ~[na:1.8.0_101]
    at org.springframework.boot.devtools.restart.RestartLauncher.run(RestartLauncher.java:49) [spring-boot-devtools-2.1.0.RELEASE.jar:2.1.0.RELEASE]
Caused by: org.springframework.beans.factory.NoSuchBeanDefinitionException: No qualifying bean of type 'org.springframework.kafka.core.ConsumerFactory<java.lang.Object, java.lang.Object>' available: expected at least 1 bean which qualifies as autowire candidate. Dependency annotations: {}
    at org.springframework.beans.factory.support.DefaultListableBeanFactory.raiseNoMatchingBeanFound(DefaultListableBeanFactory.java:1646) ~[spring-beans-5.1.2.RELEASE.jar:5.1.2.RELEASE]
    at org.springframework.beans.factory.support.DefaultListableBeanFactory.doResolveDependency(DefaultListableBeanFactory.java:1205) ~[spring-beans-5.1.2.RELEASE.jar:5.1.2.RELEASE]
    at org.springframework.beans.factory.support.DefaultListableBeanFactory.resolveDependency(DefaultListableBeanFactory.java:1166) ~[spring-beans-5.1.2.RELEASE.jar:5.1.2.RELEASE]
    at org.springframework.beans.factory.support.ConstructorResolver.resolveAutowiredArgument(ConstructorResolver.java:855) ~[spring-beans-5.1.2.RELEASE.jar:5.1.2.RELEASE]
    at org.springframework.beans.factory.support.ConstructorResolver.createArgumentArray(ConstructorResolver.java:758) ~[spring-beans-5.1.2.RELEASE.jar:5.1.2.RELEASE]
    ... 24 common frames omitted

    :

    正如回答中指出的,这个问题是由于消费者工厂的定制造成的。使用自动配置解决了该问题。

    :我可以通过创建ConsumerFactory类型的bean来避免这个问题,但我只是想知道它是否可以作为自动配置本身的一部分进行控制

    1 回复  |  直到 7 年前
        1
  •  2
  •   Gary Russell    7 年前

    您需要显示所有配置(pf、cf等)。

    然而,您似乎混淆了容器和工厂。

    KafkaMessageListenerContainer

    也就是说,我认为你把事情复杂化了——只需使用Boot的自动配置工厂来创建回复容器。这对我来说很好。。。 

    public class So53573729Application {

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

    @Bean
    public ConcurrentMessageListenerContainer<Long, String> repliesContainer(
            ConcurrentKafkaListenerContainerFactory<Long, String> containerFactory) {

        ConcurrentMessageListenerContainer<Long, String> repliesContainer = 
                containerFactory.createContainer("replies");
        repliesContainer.getContainerProperties().setGroupId("repliesGroup");
        repliesContainer.setAutoStartup(false);
        return repliesContainer;
    }

    @Bean
    public ReplyingKafkaTemplate<Long, String, String> replyingTemplate(
            ProducerFactory<Long, String> pf,
            ConcurrentMessageListenerContainer<Long, String> repliesContainer) {

        return new ReplyingKafkaTemplate<>(pf, repliesContainer);
    }

}
    推荐文章
    user25089203  ·  替代Kafka 3.6.2中的@EnableBinding和@StreamListener
    2 年前
    springenthusiast  ·  如何将springboot中的重复配置作为jar并用作依赖项
    2 年前
    bohunn  ·  Kafka Streams-无法搜索状态存储
    3 年前
    Andrey P.  ·  添加spring-cloud-starter-bus-Kafka依赖项后,Kafka消费者没有收到消息
    3 年前
    StevenPG  ·  Spring Cloud Streams错误处理不起作用
    3 年前
    pankaj  ·  Kafka中的电流偏移量怎么会这么高,而日志端偏移量太低,导致负Kafka滞后过高?
    3 年前
    关于   移动版
    代码之家 - 一站式码农服务社区
    沪ICP备11025650号