Spring 集成 Kafka

Spring Integration Kafka

我正在尝试从我的 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.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$45d892.producer(<generated>)
    at demo.IntegrationConfiguration.config(IntegrationConfiguration.java:178)
    at demo.IntegrationConfiguration$$EnhancerBySpringCGLIB$45d892.CGLIB$config[=13=](<generated>)
    at demo.IntegrationConfiguration$$EnhancerBySpringCGLIB$45d892$$FastClassBySpringCGLIB$e71af6.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$45d892.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 分发进行了尝试,异常仍然存在

尝试更改

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

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

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