我用的spring-cloud-stream不能发消息，但是可以接收消息
I used spring-cloud-stream can not send message, but can receive message
StreamInput.java
public interface StreamInput {
String QUEUE = "cherryQueue";
@Input(StreamInput.QUEUE)
SubscribableChannel input();
}
StreamOutput.java
public interface StreamOutput {
@Output(StreamInput.QUEUE)
MessageChannel output();
}
StreamReceiver.java
@Slf4j
@Component
@EnableBinding(StreamInput.class)
public class StreamReceiver {
@StreamListener(StreamInput.QUEUE)
public void process(Object message){
log.info("receive message:{}",message);
log.info("--------------------------------------------------");
}
}
springboot项目启动成功，队列创建成功：
enter image description here
============================================= ===========================
但是！我用的是StreamOutput，不行。
这是测试用例：
@Slf4j
@RunWith(SpringRunner.class)
@SpringBootTest
@EnableBinding(StreamOutput.class)
public class MessageTest {
@Autowired
private StreamOutput streamOutput;
@Test
public void sentMessageTest() {
streamOutput.output()
.send(MessageBuilder.withPayload("now time is :" + new Date()).build());
log.info("the message send OK !!!");
}
}
这是异常信息：
java.lang.IllegalStateException: Failed to load ApplicationContext
at org.springframework.test.context.cache.DefaultCacheAwareContextLoaderDelegate.loadContext(DefaultCacheAwareContextLoaderDelegate.java:125)
at org.springframework.test.context.support.DefaultTestContext.getApplicationContext(DefaultTestContext.java:108)
at org.springframework.test.context.web.ServletTestExecutionListener.setUpRequestContextIfNecessary(ServletTestExecutionListener.java:190)
at org.springframework.test.context.web.ServletTestExecutionListener.prepareTestInstance(ServletTestExecutionListener.java:132)
at org.springframework.test.context.TestContextManager.prepareTestInstance(TestContextManager.java:246)
at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.createTest(SpringJUnit4ClassRunner.java:227)
at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.runReflectiveCall(SpringJUnit4ClassRunner.java:289)
at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.methodBlock(SpringJUnit4ClassRunner.java:291)
at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.runChild(SpringJUnit4ClassRunner.java:246)
at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.runChild(SpringJUnit4ClassRunner.java:97)
at org.junit.runners.ParentRunner.run(ParentRunner.java:290)
at org.junit.runners.ParentRunner.schedule(ParentRunner.java:71)
at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288)
at org.junit.runners.ParentRunner.access[=15=]0(ParentRunner.java:58)
at org.junit.runners.ParentRunner.evaluate(ParentRunner.java:268)
at org.springframework.test.context.junit4.statements.RunBeforeTestClassCallbacks.evaluate(RunBeforeTestClassCallbacks.java:61)
at org.springframework.test.context.junit4.statements.RunAfterTestClassCallbacks.evaluate(RunAfterTestClassCallbacks.java:70)
at org.junit.runners.ParentRunner.run(ParentRunner.java:363)
at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.run(SpringJUnit4ClassRunner.java:190)
at org.junit.runner.JUnitCore.run(JUnitCore.java:137)
at com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:68)
at com.intellij.rt.execution.junit.IdeaTestRunner$Repeater.startRunnerWithArgs(IdeaTestRunner.java:47)
at com.intellij.rt.execution.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:242)
at com.intellij.rt.execution.junit.JUnitStarter.main(JUnitStarter.java:70)
Caused by: org.springframework.beans.factory.BeanDefinitionStoreException: Invalid bean definition with name 'cherryQueue' defined in com.cherry.order.message.StreamInput: bean definition with this name already exists - Root bean: class [null]; scope=; abstract=false; lazyInit=false; autowireMode=0; dependencyCheck=0; autowireCandidate=true; primary=false; factoryBeanName=com.cherry.order.message.MessageTest$StreamOutput; factoryMethodName=output; initMethodName=null; destroyMethodName=null
at org.springframework.cloud.stream.binding.BindingBeanDefinitionRegistryUtils.registerBindingTargetBeanDefinition(BindingBeanDefinitionRegistryUtils.java:64)
at org.springframework.cloud.stream.binding.BindingBeanDefinitionRegistryUtils.registerInputBindingTargetBeanDefinition(BindingBeanDefinitionRegistryUtils.java:47)
at org.springframework.cloud.stream.binding.BindingBeanDefinitionRegistryUtils.lambda$registerBindingTargetBeanDefinitions[=15=](BindingBeanDefinitionRegistryUtils.java:80)
at org.springframework.util.ReflectionUtils.doWithMethods(ReflectionUtils.java:562)
at org.springframework.util.ReflectionUtils.doWithMethods(ReflectionUtils.java:541)
at org.springframework.cloud.stream.binding.BindingBeanDefinitionRegistryUtils.registerBindingTargetBeanDefinitions(BindingBeanDefinitionRegistryUtils.java:76)
at org.springframework.cloud.stream.config.BindingBeansRegistrar.registerBeanDefinitions(BindingBeansRegistrar.java:45)
at org.springframework.context.annotation.ConfigurationClassBeanDefinitionReader.lambda$loadBeanDefinitionsFromRegistrars(ConfigurationClassBeanDefinitionReader.java:358)
at java.util.LinkedHashMap.forEach(LinkedHashMap.java:684)
at org.springframework.context.annotation.ConfigurationClassBeanDefinitionReader.loadBeanDefinitionsFromRegistrars(ConfigurationClassBeanDefinitionReader.java:357)
at org.springframework.context.annotation.ConfigurationClassBeanDefinitionReader.loadBeanDefinitionsForConfigurationClass(ConfigurationClassBeanDefinitionReader.java:145)
at org.springframework.context.annotation.ConfigurationClassBeanDefinitionReader.loadBeanDefinitions(ConfigurationClassBeanDefinitionReader.java:117)
at org.springframework.context.annotation.ConfigurationClassPostProcessor.processConfigBeanDefinitions(ConfigurationClassPostProcessor.java:328)
at org.springframework.context.annotation.ConfigurationClassPostProcessor.postProcessBeanDefinitionRegistry(ConfigurationClassPostProcessor.java:233)
at org.springframework.context.support.PostProcessorRegistrationDelegate.invokeBeanDefinitionRegistryPostProcessors(PostProcessorRegistrationDelegate.java:273)
at org.springframework.context.support.PostProcessorRegistrationDelegate.invokeBeanFactoryPostProcessors(PostProcessorRegistrationDelegate.java:93)
at org.springframework.context.support.AbstractApplicationContext.invokeBeanFactoryPostProcessors(AbstractApplicationContext.java:694)
at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:532)
at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:759)
at org.springframework.boot.SpringApplication.refreshContext(SpringApplication.java:395)
at org.springframework.boot.SpringApplication.run(SpringApplication.java:327)
at org.springframework.boot.test.context.SpringBootContextLoader.loadContext(SpringBootContextLoader.java:139)
at org.springframework.test.context.cache.DefaultCacheAwareContextLoaderDelegate.loadContextInternal(DefaultCacheAwareContextLoaderDelegate.java:99)
at org.springframework.test.context.cache.DefaultCacheAwareContextLoaderDelegate.loadContext(DefaultCacheAwareContextLoaderDelegate.java:117)
... 24 more
Caused by: org.springframework.beans.factory.BeanDefinitionStoreException: Invalid bean definition with name 'cherryQueue' defined in com.cherry.order.message.StreamInput: bean definition with this name already exists - Root bean: class [null]; scope=; abstract=false; lazyInit=false; autowireMode=0; dependencyCheck=0; autowireCandidate=true; primary=false; factoryBeanName=com.cherry.order.message.MessageTest$StreamOutput; factoryMethodName=output; initMethodName=null; destroyMethodName=null
我觉得错误很明显。
@Output(StreamInput.QUEUE)
您不能为 2 个频道使用相同的频道名称。
如果您希望在同一应用程序中生成和使用来自同一目标的消息（这有点不寻常），您必须使用 2 个通道，并将绑定配置为引用同一目标。
为什么这种情况不常见，我确实计划将两个微服务最初开发为一个应用程序以加快开发速度并使用消息进行通信。
一切都在使用纯 Rabbit，现在当我切换到 Spring Cloud Stream 时它停止了。
我在另一个微服务中还有另一个 'real' 接收器。
使用不同的 'bean names' 和覆盖目标定义接收器和源仍然不起作用，因为我怀疑 Spring 使用相同线程（DirectChannel）将来自源的消息转发到内部接收器（跳过兔子！） .
StreamInput.java
public interface StreamInput {
String QUEUE = "cherryQueue";
@Input(StreamInput.QUEUE)
SubscribableChannel input();
}
StreamOutput.java
public interface StreamOutput {
@Output(StreamInput.QUEUE)
MessageChannel output();
}
StreamReceiver.java
@Slf4j
@Component
@EnableBinding(StreamInput.class)
public class StreamReceiver {
@StreamListener(StreamInput.QUEUE)
public void process(Object message){
log.info("receive message:{}",message);
log.info("--------------------------------------------------");
}
}
springboot项目启动成功，队列创建成功： enter image description here
============================================= =========================== 但是！我用的是StreamOutput，不行。
这是测试用例：
@Slf4j
@RunWith(SpringRunner.class)
@SpringBootTest
@EnableBinding(StreamOutput.class)
public class MessageTest {
@Autowired
private StreamOutput streamOutput;
@Test
public void sentMessageTest() {
streamOutput.output()
.send(MessageBuilder.withPayload("now time is :" + new Date()).build());
log.info("the message send OK !!!");
}
}
这是异常信息：
java.lang.IllegalStateException: Failed to load ApplicationContext
at org.springframework.test.context.cache.DefaultCacheAwareContextLoaderDelegate.loadContext(DefaultCacheAwareContextLoaderDelegate.java:125)
at org.springframework.test.context.support.DefaultTestContext.getApplicationContext(DefaultTestContext.java:108)
at org.springframework.test.context.web.ServletTestExecutionListener.setUpRequestContextIfNecessary(ServletTestExecutionListener.java:190)
at org.springframework.test.context.web.ServletTestExecutionListener.prepareTestInstance(ServletTestExecutionListener.java:132)
at org.springframework.test.context.TestContextManager.prepareTestInstance(TestContextManager.java:246)
at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.createTest(SpringJUnit4ClassRunner.java:227)
at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.runReflectiveCall(SpringJUnit4ClassRunner.java:289)
at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.methodBlock(SpringJUnit4ClassRunner.java:291)
at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.runChild(SpringJUnit4ClassRunner.java:246)
at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.runChild(SpringJUnit4ClassRunner.java:97)
at org.junit.runners.ParentRunner.run(ParentRunner.java:290)
at org.junit.runners.ParentRunner.schedule(ParentRunner.java:71)
at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288)
at org.junit.runners.ParentRunner.access[=15=]0(ParentRunner.java:58)
at org.junit.runners.ParentRunner.evaluate(ParentRunner.java:268)
at org.springframework.test.context.junit4.statements.RunBeforeTestClassCallbacks.evaluate(RunBeforeTestClassCallbacks.java:61)
at org.springframework.test.context.junit4.statements.RunAfterTestClassCallbacks.evaluate(RunAfterTestClassCallbacks.java:70)
at org.junit.runners.ParentRunner.run(ParentRunner.java:363)
at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.run(SpringJUnit4ClassRunner.java:190)
at org.junit.runner.JUnitCore.run(JUnitCore.java:137)
at com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:68)
at com.intellij.rt.execution.junit.IdeaTestRunner$Repeater.startRunnerWithArgs(IdeaTestRunner.java:47)
at com.intellij.rt.execution.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:242)
at com.intellij.rt.execution.junit.JUnitStarter.main(JUnitStarter.java:70)
Caused by: org.springframework.beans.factory.BeanDefinitionStoreException: Invalid bean definition with name 'cherryQueue' defined in com.cherry.order.message.StreamInput: bean definition with this name already exists - Root bean: class [null]; scope=; abstract=false; lazyInit=false; autowireMode=0; dependencyCheck=0; autowireCandidate=true; primary=false; factoryBeanName=com.cherry.order.message.MessageTest$StreamOutput; factoryMethodName=output; initMethodName=null; destroyMethodName=null
at org.springframework.cloud.stream.binding.BindingBeanDefinitionRegistryUtils.registerBindingTargetBeanDefinition(BindingBeanDefinitionRegistryUtils.java:64)
at org.springframework.cloud.stream.binding.BindingBeanDefinitionRegistryUtils.registerInputBindingTargetBeanDefinition(BindingBeanDefinitionRegistryUtils.java:47)
at org.springframework.cloud.stream.binding.BindingBeanDefinitionRegistryUtils.lambda$registerBindingTargetBeanDefinitions[=15=](BindingBeanDefinitionRegistryUtils.java:80)
at org.springframework.util.ReflectionUtils.doWithMethods(ReflectionUtils.java:562)
at org.springframework.util.ReflectionUtils.doWithMethods(ReflectionUtils.java:541)
at org.springframework.cloud.stream.binding.BindingBeanDefinitionRegistryUtils.registerBindingTargetBeanDefinitions(BindingBeanDefinitionRegistryUtils.java:76)
at org.springframework.cloud.stream.config.BindingBeansRegistrar.registerBeanDefinitions(BindingBeansRegistrar.java:45)
at org.springframework.context.annotation.ConfigurationClassBeanDefinitionReader.lambda$loadBeanDefinitionsFromRegistrars(ConfigurationClassBeanDefinitionReader.java:358)
at java.util.LinkedHashMap.forEach(LinkedHashMap.java:684)
at org.springframework.context.annotation.ConfigurationClassBeanDefinitionReader.loadBeanDefinitionsFromRegistrars(ConfigurationClassBeanDefinitionReader.java:357)
at org.springframework.context.annotation.ConfigurationClassBeanDefinitionReader.loadBeanDefinitionsForConfigurationClass(ConfigurationClassBeanDefinitionReader.java:145)
at org.springframework.context.annotation.ConfigurationClassBeanDefinitionReader.loadBeanDefinitions(ConfigurationClassBeanDefinitionReader.java:117)
at org.springframework.context.annotation.ConfigurationClassPostProcessor.processConfigBeanDefinitions(ConfigurationClassPostProcessor.java:328)
at org.springframework.context.annotation.ConfigurationClassPostProcessor.postProcessBeanDefinitionRegistry(ConfigurationClassPostProcessor.java:233)
at org.springframework.context.support.PostProcessorRegistrationDelegate.invokeBeanDefinitionRegistryPostProcessors(PostProcessorRegistrationDelegate.java:273)
at org.springframework.context.support.PostProcessorRegistrationDelegate.invokeBeanFactoryPostProcessors(PostProcessorRegistrationDelegate.java:93)
at org.springframework.context.support.AbstractApplicationContext.invokeBeanFactoryPostProcessors(AbstractApplicationContext.java:694)
at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:532)
at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:759)
at org.springframework.boot.SpringApplication.refreshContext(SpringApplication.java:395)
at org.springframework.boot.SpringApplication.run(SpringApplication.java:327)
at org.springframework.boot.test.context.SpringBootContextLoader.loadContext(SpringBootContextLoader.java:139)
at org.springframework.test.context.cache.DefaultCacheAwareContextLoaderDelegate.loadContextInternal(DefaultCacheAwareContextLoaderDelegate.java:99)
at org.springframework.test.context.cache.DefaultCacheAwareContextLoaderDelegate.loadContext(DefaultCacheAwareContextLoaderDelegate.java:117)
... 24 more
Caused by: org.springframework.beans.factory.BeanDefinitionStoreException: Invalid bean definition with name 'cherryQueue' defined in com.cherry.order.message.StreamInput: bean definition with this name already exists - Root bean: class [null]; scope=; abstract=false; lazyInit=false; autowireMode=0; dependencyCheck=0; autowireCandidate=true; primary=false; factoryBeanName=com.cherry.order.message.MessageTest$StreamOutput; factoryMethodName=output; initMethodName=null; destroyMethodName=null
我觉得错误很明显。
@Output(StreamInput.QUEUE)
您不能为 2 个频道使用相同的频道名称。
如果您希望在同一应用程序中生成和使用来自同一目标的消息（这有点不寻常），您必须使用 2 个通道，并将绑定配置为引用同一目标。
为什么这种情况不常见，我确实计划将两个微服务最初开发为一个应用程序以加快开发速度并使用消息进行通信。
一切都在使用纯 Rabbit，现在当我切换到 Spring Cloud Stream 时它停止了。
我在另一个微服务中还有另一个 'real' 接收器。
使用不同的 'bean names' 和覆盖目标定义接收器和源仍然不起作用，因为我怀疑 Spring 使用相同线程（DirectChannel）将来自源的消息转发到内部接收器（跳过兔子！） .