java - Spring集成Kafka

Question

我正在尝试从我的 Spring 集成应用程序解析发送给 kafka 的消息。我很确定我应该将 KafkaProducerMessageHandler 设置为我的端点，如下所示

    @Bean
    @ServiceActivator(inputChannel = "correctChannel")
    public KafkaProducerMessageHandler<String, String> handler() throws Exception{
        return new KafkaProducerMessageHandler<String, String>(kafkaContext());
    } 

然后我尝试构建构造函数的所有参数，如下所示

@Bean
public KafkaProducerContext<String, String> kafkaContext () throws Exception{
    KafkaProducerContext<String, String> ctx = new KafkaProducerContext<String, String>();
    ctx.setProducerConfigurations(Collections.singletonMap("test", config()));
    return ctx;
}

@Bean
public ProducerConfiguration<String, String> config() throws Exception{
    return new ProducerConfiguration<String, String>(metaData(), producer().getObject());
}

@Bean
public Encoder<String> encoder(){
    return new StringEncoder<String>();
}

@Bean
public ProducerMetadata<String, String> metaData(){
    ProducerMetadata<String, String> meta = new ProducerMetadata<String,String>("test");
    meta.setValueClassType(String.class);
    meta.setKeyClassType(String.class);
    meta.setValueEncoder(encoder());
    meta.setKeyEncoder(encoder());
    return meta;
}

@Bean
public ProducerFactoryBean<String, String> producer(){
    return new ProducerFactoryBean<String,String>(metaData(), "localhost:9092");
}

但是 ProducerConfiguration 部分抛出此异常:

Exception in thread "main" org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'config' defined in class path resource [demo/IntegrationConfiguration.class]: Bean instantiation via factory method failed; nested exception is org.springframework.beans.BeanInstantiationException: Failed to instantiate [org.springframework.integration.kafka.support.ProducerConfiguration]: Factory method 'config' threw exception; nested exception is java.lang.IllegalArgumentException: Superclass has no null constructors but no arguments were given
    at org.springframework.beans.factory.support.ConstructorResolver.instantiateUsingFactoryMethod(ConstructorResolver.java:599)
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.instantiateUsingFactoryMethod(AbstractAutowireCapableBeanFactory.java:1111)
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.createBeanInstance(AbstractAutowireCapableBeanFactory.java:1006)
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.doCreateBean(AbstractAutowireCapableBeanFactory.java:504)
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.createBean(AbstractAutowireCapableBeanFactory.java:476)
    at org.springframework.beans.factory.support.AbstractBeanFactory$1.getObject(AbstractBeanFactory.java:303)
    at org.springframework.beans.factory.support.DefaultSingletonBeanRegistry.getSingleton(DefaultSingletonBeanRegistry.java:230)
    at org.springframework.beans.factory.support.AbstractBeanFactory.doGetBean(AbstractBeanFactory.java:299)
    at org.springframework.beans.factory.support.AbstractBeanFactory.getBean(AbstractBeanFactory.java:194)
    at org.springframework.beans.factory.support.DefaultListableBeanFactory.preInstantiateSingletons(DefaultListableBeanFactory.java:755)
    at org.springframework.context.support.AbstractApplicationContext.finishBeanFactoryInitialization(AbstractApplicationContext.java:757)
    at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:480)
    at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:686)
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:320)
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:957)
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:946)
    at demo.IntegrationnnnApplication.main(IntegrationnnnApplication.java:12)
Caused by: org.springframework.beans.BeanInstantiationException: Failed to instantiate [org.springframework.integration.kafka.support.ProducerConfiguration]: Factory method 'config' threw exception; nested exception is java.lang.IllegalArgumentException: Superclass has no null constructors but no arguments were given
    at org.springframework.beans.factory.support.SimpleInstantiationStrategy.instantiate(SimpleInstantiationStrategy.java:189)
    at org.springframework.beans.factory.support.ConstructorResolver.instantiateUsingFactoryMethod(ConstructorResolver.java:588)
    ... 16 more
Caused by: java.lang.IllegalArgumentException: Superclass has no null constructors but no arguments were given
    at org.springframework.cglib.proxy.Enhancer.emitConstructors(Enhancer.java:721)
    at org.springframework.cglib.proxy.Enhancer.generateClass(Enhancer.java:499)
    at org.springframework.cglib.core.DefaultGeneratorStrategy.generate(DefaultGeneratorStrategy.java:25)
    at org.springframework.cglib.core.AbstractClassGenerator.create(AbstractClassGenerator.java:216)
    at org.springframework.cglib.proxy.Enhancer.createHelper(Enhancer.java:377)
    at org.springframework.cglib.proxy.Enhancer.create(Enhancer.java:285)
    at org.springframework.context.annotation.ConfigurationClassEnhancer$BeanMethodInterceptor.enhanceFactoryBean(ConfigurationClassEnhancer.java:384)
    at org.springframework.context.annotation.ConfigurationClassEnhancer$BeanMethodInterceptor.intercept(ConfigurationClassEnhancer.java:292)
    at demo.IntegrationConfiguration$$EnhancerBySpringCGLIB$$8445d892.producer(<generated>)
    at demo.IntegrationConfiguration.config(IntegrationConfiguration.java:178)
    at demo.IntegrationConfiguration$$EnhancerBySpringCGLIB$$8445d892.CGLIB$config$0(<generated>)
    at demo.IntegrationConfiguration$$EnhancerBySpringCGLIB$$8445d892$$FastClassBySpringCGLIB$$59e71af6.invoke(<generated>)
    at org.springframework.cglib.proxy.MethodProxy.invokeSuper(MethodProxy.java:228)
    at org.springframework.context.annotation.ConfigurationClassEnhancer$BeanMethodInterceptor.intercept(ConfigurationClassEnhancer.java:309)
    at demo.IntegrationConfiguration$$EnhancerBySpringCGLIB$$8445d892.config(<generated>)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:483)
    at org.springframework.beans.factory.support.SimpleInstantiationStrategy.instantiate(SimpleInstantiationStrategy.java:162)
    ... 17 more

我不知道这是什么意思。

另一个问题:是否可以向 kafka 发送具有多个负载的消息？

编辑:按照 spring-integration-kafka 文档中所述，使用 Scala 版本 2.9.2 和 2.9.2 的正确 kafka 发行版进行了尝试，异常仍然存在

Answer 1

尝试改变

@Bean
public ProducerFactoryBean<String, String> producer(){ 

到

@Bean
public Producer<String, String> producer() { 

并删除producer().getObject()中的.getObject()。换句话说，让框架负责调用工厂 bean 来获取 bean 实例。