spring - 已通过 Spring Cloud Stream 提供 RabbitMq 的生产者
问题描述
我发现自己处于一个已经将rabbitmq 作为给定基础设施提供的环境中。
服务 A正在写入兔子队列,服务 B正在从队列中读取。通过Spring Cloud Stream Binder作为消费者的阅读部分在大量阅读文档并正确设置后工作得很好。
但是,我无法设置要写入兔子队列的生产者。
设置
RabbitMq(已经启动并运行)
- 1个交易所:myExchange
- 3 个队列:myQueueA、myQueueB、myQueueC(在myExchange下链接)
Producer-Service(上面提到的 Service A)
@Slf4j
@Component
@RequiredArgsConstructor
@EnableBinding(RabbitChannelSource.class)
public class RabbitSender
{
private final RabbitChannelSource rabbitChannelSource;
public void sendMessage()
{
Message<String> msg = MessageBuilder.withPayload("TEEEEST").build();
rabbitChannelSource.myOutput().send(msg);
}
public interface RabbitChannelSource
{
String MY_OUTPUT_BINDING = "my-output";
@Output(MY_OUTPUT_BINDING)
MessageChannel myOutput();
}
我试图让它首先为一个队列工作,但理想情况下我会为所有三个队列设置正确的属性。
spring.rabbitmq.host=localhost
spring.rabbitmq.port=5672
spring.rabbitmq.username=user
spring.rabbitmq.password=pass
## use existing rabbitmq via bindings
spring.cloud.stream.bindings.my-output.destination=my-output
#spring.cloud.stream.bindings.my-output.group=myQueueA
spring.cloud.stream.bindings.my-output.producer.required-groups=myQueueA
spring.cloud.stream.rabbit.bindings.my-output.producer.bind-queue=false
spring.cloud.stream.rabbit.bindings.my-output.producer.declare-exchange=false
spring.cloud.stream.rabbit.bindings.my-output.producer.queueNameGroupOnly=true
bind-queue=false
并且declare-exchange=false
是必要的,因为我有兔子基础设施。
但是,但是我总是遇到同样的异常,我不知道为什么。我的意思是我知道为什么,因为没有合适的通道可以发送消息。所以我怀疑它与application.properties
.
org.springframework.messaging.MessageDeliveryException: Dispatcher has no subscribers for channel 'application.my-output'.; nested exception is org.springframework.integration.MessageDispatchingException: Dispatcher has no subscribers, failedMessage=GenericMessage [payload=byte[7], headers={contentType=application/json, id=98698b4c-61fa-596d-736e-f630d3ba4626, timestamp=1605105272723}]
at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:77)
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:453)
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:403)
at de.techem.emsreceiver.rabbitmq.RabbitSender.sendMessage(RabbitSender.java:25)
at de.techem.emsreceiver.event.TriggeredEmsImport.execute(TriggeredImport.java:95)
at de.techem.emsreceiver.event.TriggeredEmsImport$$FastClassBySpringCGLIB$$d84f31a4.invoke(<generated>)
at org.springframework.cglib.proxy.MethodProxy.invoke(MethodProxy.java:218)
at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.invokeJoinpoint(CglibAopProxy.java:769)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163)
at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.proceed(CglibAopProxy.java:747)
at org.springframework.aop.aspectj.AspectJAfterThrowingAdvice.invoke(AspectJAfterThrowingAdvice.java:62)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:175)
at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.proceed(CglibAopProxy.java:747)
at org.springframework.aop.interceptor.ExposeInvocationInterceptor.invoke(ExposeInvocationInterceptor.java:95)
at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186)
at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.proceed(CglibAopProxy.java:747)
at org.springframework.aop.framework.CglibAopProxy$DynamicAdvisedInterceptor.intercept(CglibAopProxy.java:689)
at de.techem.emsreceiver.event.TriggeredEmsImport$$EnhancerBySpringCGLIB$$a0878aed.execute(<generated>)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.base/java.lang.reflect.Method.invoke(Method.java:566)
at org.springframework.context.event.ApplicationListenerMethodAdapter.doInvoke(ApplicationListenerMethodAdapter.java:305)
at org.springframework.context.event.ApplicationListenerMethodAdapter.processEvent(ApplicationListenerMethodAdapter.java:190)
at org.springframework.context.event.ApplicationListenerMethodAdapter.onApplicationEvent(ApplicationListenerMethodAdapter.java:153)
at org.springframework.context.event.SimpleApplicationEventMulticaster.doInvokeListener(SimpleApplicationEventMulticaster.java:172)
at org.springframework.context.event.SimpleApplicationEventMulticaster.invokeListener(SimpleApplicationEventMulticaster.java:165)
at org.springframework.context.event.SimpleApplicationEventMulticaster.multicastEvent(SimpleApplicationEventMulticaster.java:139)
at org.springframework.context.support.AbstractApplicationContext.publishEvent(AbstractApplicationContext.java:403)
at org.springframework.context.support.AbstractApplicationContext.publishEvent(AbstractApplicationContext.java:409)
at org.springframework.context.support.AbstractApplicationContext.publishEvent(AbstractApplicationContext.java:360)
at org.springframework.boot.context.event.EventPublishingRunListener.running(EventPublishingRunListener.java:103)
at org.springframework.boot.SpringApplicationRunListeners.running(SpringApplicationRunListeners.java:77)
at org.springframework.boot.SpringApplication.run(SpringApplication.java:330)
at org.springframework.boot.builder.SpringApplicationBuilder.run(SpringApplicationBuilder.java:140)
at org.springframework.cloud.stream.binder.DefaultBinderFactory.getBinderInstance(DefaultBinderFactory.java:320)
at org.springframework.cloud.stream.binder.DefaultBinderFactory.doGetBinder(DefaultBinderFactory.java:209)
at org.springframework.cloud.stream.binder.DefaultBinderFactory.getBinder(DefaultBinderFactory.java:140)
at org.springframework.cloud.stream.binding.BindingService.getBinder(BindingService.java:379)
at org.springframework.cloud.stream.binding.BindingService.bindProducer(BindingService.java:268)
at org.springframework.cloud.stream.binding.BindingService.bindProducer(BindingService.java:291)
at org.springframework.cloud.stream.binding.AbstractBindableProxyFactory.createAndBindOutputs(AbstractBindableProxyFactory.java:136)
at org.springframework.cloud.stream.binding.OutputBindingLifecycle.doStartWithBindable(OutputBindingLifecycle.java:58)
at java.base/java.util.LinkedHashMap$LinkedValues.forEach(LinkedHashMap.java:608)
at org.springframework.cloud.stream.binding.AbstractBindingLifecycle.start(AbstractBindingLifecycle.java:57)
at org.springframework.cloud.stream.binding.OutputBindingLifecycle.start(OutputBindingLifecycle.java:34)
at org.springframework.context.support.DefaultLifecycleProcessor.doStart(DefaultLifecycleProcessor.java:182)
at org.springframework.context.support.DefaultLifecycleProcessor.access$200(DefaultLifecycleProcessor.java:53)
at org.springframework.context.support.DefaultLifecycleProcessor$LifecycleGroup.start(DefaultLifecycleProcessor.java:360)
at org.springframework.context.support.DefaultLifecycleProcessor.startBeans(DefaultLifecycleProcessor.java:158)
at org.springframework.context.support.DefaultLifecycleProcessor.onRefresh(DefaultLifecycleProcessor.java:122)
at org.springframework.context.support.AbstractApplicationContext.finishRefresh(AbstractApplicationContext.java:894)
at org.springframework.boot.web.servlet.context.ServletWebServerApplicationContext.finishRefresh(ServletWebServerApplicationContext.java:162)
at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:553)
at org.springframework.boot.web.servlet.context.ServletWebServerApplicationContext.refresh(ServletWebServerApplicationContext.java:141)
at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:747)
at org.springframework.boot.SpringApplication.refreshContext(SpringApplication.java:397)
at org.springframework.boot.SpringApplication.run(SpringApplication.java:315)
at org.springframework.boot.SpringApplication.run(SpringApplication.java:1226)
at org.springframework.boot.SpringApplication.run(SpringApplication.java:1215)
at de.techem.emsreceiver.EmsReceiverApplication.main(EmsReceiverApplication.java:21)
我希望能够myQueueA, myQueueB or myQueueC
基于 routingKey 写入。因此,如果我在我的应用程序中设置 routingKey 到myQA
消息应该发送到myQueueA
,myQB
然后到myQueueB
,myQC
然后到myQueueC
。
所以三个不同的路由键会导致对应的兔子队列。
我很高兴收到任何意见,因为我尝试了很多没有让我成功的事情。谢谢!
解决方案
也许您正在尝试在绑定之前使用绑定?这对我来说很好:
spring.cloud.stream.bindings.output.destination=my-output
spring.cloud.stream.rabbit.bindings.output.producer.declare-exchange=false
spring.cloud.stream.rabbit.bindings.output.producer.routing-key-expression=headers['routeTo']
@SpringBootApplication
@EnableBinding(Source.class)
public class So64788954Application {
public static void main(String[] args) {
SpringApplication.run(So64788954Application.class, args);
}
@Autowired
Source source;
@Bean
ApplicationRunner runner() {
return args -> {
source.output().send(MessageBuilder.withPayload("foo")
.setHeader("routeTo", "myQA")
.build());
source.output().send(MessageBuilder.withPayload("bar")
.setHeader("routeTo", "myQB")
.build());
source.output().send(MessageBuilder.withPayload("baz")
.setHeader("routeTo", "myQC")
.build());
};
}
@RabbitListener(queues = { "myQueueA", "myQueueB", "myQueueC" } )
void listen(String in, @Header(AmqpHeaders.CONSUMER_QUEUE) String queue) {
System.out.println(in + ", from: " + queue);
}
}
foo, from: myQueueA
bar, from: myQueueB
baz, from: myQueueC
当您带上自己的队列/交换时,您不需要生产者端所需的组。
The
@RabbitListener` 只是用来消费发送到 3 个队列的消息。
顺便说一句,注释模型已被弃用;现在首选函数式编程模型,我们使用StreamBridge
for 输出(和Consumer<?>
或Function<?, ?>
分别使用和处理)。
这是上面的等价物:
spring.cloud.stream.rabbit.bindings.my-output.producer.declare-exchange=false
spring.cloud.stream.rabbit.bindings.my-output.producer.routing-key-expression=headers['routeTo']
@SpringBootApplication
public class So64788954Application {
public static void main(String[] args) {
SpringApplication.run(So64788954Application.class, args);
}
@Autowired
StreamBridge bridge;
@Bean
ApplicationRunner runner() {
return args -> {
bridge.send("my-output", MessageBuilder.withPayload("foo")
.setHeader("routeTo", "myQA")
.build());
bridge.send("my-output", MessageBuilder.withPayload("bar")
.setHeader("routeTo", "myQB")
.build());
bridge.send("my-output", MessageBuilder.withPayload("baz")
.setHeader("routeTo", "myQC")
.build());
};
}
@RabbitListener(queues = { "myQueueA", "myQueueB", "myQueueC" } )
void listen(String in, @Header(AmqpHeaders.CONSUMER_QUEUE) String queue) {
System.out.println(in + ", from: " + queue);
}
}
推荐阅读
- html - 如何从导航栏中单击锚链接并在没有 javascript 的情况下更改 div 中的内容?
- mysql - 每周查询新用户数 - 如何按年/周排序?
- windows - 建立外部连接以在 Windows 上的 docker 映像上安装 git 时出错
- azure - 在管理组中部署自定义 Azure 策略的选项
- angular - 使用 --prod 标志构建时 AngularFireFunctions 注入错误
- firebase - Firebase:突然无法再部署
- r - 如何在循环中使用 cbind 命名列
- java - `thenRunAsync(...)` 和 `CompletableFuture.runAsync(() -> { ... });` 有关系吗?
- flutter - 在颤动的标签栏的应用栏中添加返回按钮
- c++ - 这种快速排序算法有什么问题?