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 实例。
我正在尝试从我的 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 实例。