Spring Cloud Stream可连接到多个主机以实现单个绑定器(RabbitMQ)

4jb9z9bj 于 2022-11-08 发布在 RabbitMQ

关注(0)|答案(2)|浏览(241)

我们正在使用Spring Cloud Stream来监听rabbitMQ多个队列，尤其是SCF模型

不赞成使用spring-cloud-stream-reactive模块，而赞成使用通过Spring Cloud Function编程模型提供的本地支持。

当存在单个节点/主机时，它工作良好（下面共享的application.yml片段），
然而，当我们尝试连接多个节点时，它失败了，有人能指导如何连接相同的节点吗？或者有一些与Spring云文档相关的示例吗

以下代码按预期运行

spring:
  cloud:
    stream:
      function:
        definition: function1;function2;function3
      bindings:
        function1-in-0:
          group: allocation
          destination: destinationExchange
          binder: rabbit
        function2-in-0:
          group: allocation
          destination: destinationExchange
          binder: rabbit
        function3-in-0:
          group: allocation
          destination: destinationExchange
          binder: rabbit
      rabbit:
        bindings:
          function1-in-0:
            consumer:
              bindingRoutingKey: routing.key.1
          function2-in-0:
            consumer:
              bindingRoutingKey: routing.key.2
          function3-in-0:
            consumer:
              bindingRoutingKey: routing.key.3
        binder:
          nodes: address1

基本上应该像下面这样

spring:
  cloud:
    stream:
      function:
        definition: function1;function2;function3
      bindings:
        function1-in-0:
          group: allocation
          destination: destinationExchange
          binder: rabbit1
        function2-in-0:
          group: allocation
          destination: destinationExchange
          binder: rabbit2
        function3-in-0:
          group: allocation
          destination: destinationExchange
          binder: rabbit3
      binder:
        rabbit1:
          function1-in-0:
            consumer:
              bindingRoutingKey: routing.key.1
          binder:
            nodes: address1
        rabbit2:
          function2-in-0:
            consumer:
              bindingRoutingKey: routing.key.2
          binder:
            nodes: address2
        rabbit3:
          function3-in-0:
            consumer:
              bindingRoutingKey: routing.key.3
          binder:
            nodes: address3

加上以下内容

binders:
    rabbit1:
      type: rabbit
      environment:
        spring.spring.cloud.stream.kafka:
          binder:
            nodes: localhost

我得到这个错误

o.s.boot.SpringApplication               : Application run failed

org.springframework.context.ApplicationContextException: Failed to start bean 'inputBindingLifecycle'; nested exception is java.lang.IllegalStateException: Unknown binder configuration: rabbit
    at org.springframework.context.support.DefaultLifecycleProcessor.doStart(DefaultLifecycleProcessor.java:181) ~[spring-context-5.3.8.jar:5.3.8]
    at org.springframework.context.support.DefaultLifecycleProcessor.access$200(DefaultLifecycleProcessor.java:54) ~[spring-context-5.3.8.jar:5.3.8]
    at org.springframework.context.support.DefaultLifecycleProcessor$LifecycleGroup.start(DefaultLifecycleProcessor.java:356) ~[spring-context-5.3.8.jar:5.3.8]
    at java.base/java.lang.Iterable.forEach(Iterable.java:75) ~[na:na]
    at org.springframework.context.support.DefaultLifecycleProcessor.startBeans(DefaultLifecycleProcessor.java:155) ~[spring-context-5.3.8.jar:5.3.8]
    at org.springframework.context.support.DefaultLifecycleProcessor.onRefresh(DefaultLifecycleProcessor.java:123) ~[spring-context-5.3.8.jar:5.3.8]
    at org.springframework.context.support.AbstractApplicationContext.finishRefresh(AbstractApplicationContext.java:935) ~[spring-context-5.3.8.jar:5.3.8]
    at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:586) ~[spring-context-5.3.8.jar:5.3.8]
    at org.springframework.boot.web.servlet.context.ServletWebServerApplicationContext.refresh(ServletWebServerApplicationContext.java:145) ~[spring-boot-2.5.2.jar:2.5.2]
    at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:754) ~[spring-boot-2.5.2.jar:2.5.2]
    at org.springframework.boot.SpringApplication.refreshContext(SpringApplication.java:434) ~[spring-boot-2.5.2.jar:2.5.2]
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:338) ~[spring-boot-2.5.2.jar:2.5.2]
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:1343) ~[spring-boot-2.5.2.jar:2.5.2]
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:1332) ~[spring-boot-2.5.2.jar:2.5.2]
    at com.gap.pem.Application.main(Application.java:14) ~[main/:na]
Caused by: java.lang.IllegalStateException: Unknown binder configuration: rabbit
    at org.springframework.util.Assert.state(Assert.java:76) ~[spring-core-5.3.8.jar:5.3.8]
    at org.springframework.cloud.stream.binder.DefaultBinderFactory.getBinderInstance(DefaultBinderFactory.java:255) ~[spring-cloud-stream-3.1.3.jar:3.1.3]
    at org.springframework.cloud.stream.binder.DefaultBinderFactory.doGetBinder(DefaultBinderFactory.java:224) ~[spring-cloud-stream-3.1.3.jar:3.1.3]
    at org.springframework.cloud.stream.binder.DefaultBinderFactory.getBinder(DefaultBinderFactory.java:152) ~[spring-cloud-stream-3.1.3.jar:3.1.3]
    at org.springframework.cloud.stream.binding.BindingService.getBinder(BindingService.java:386) ~[spring-cloud-stream-3.1.3.jar:3.1.3]
    at org.springframework.cloud.stream.binding.BindingService.bindConsumer(BindingService.java:103) ~[spring-cloud-stream-3.1.3.jar:3.1.3]
    at org.springframework.cloud.stream.binding.AbstractBindableProxyFactory.createAndBindInputs(AbstractBindableProxyFactory.java:118) ~[spring-cloud-stream-3.1.3.jar:3.1.3]
    at org.springframework.cloud.stream.binding.InputBindingLifecycle.doStartWithBindable(InputBindingLifecycle.java:58) ~[spring-cloud-stream-3.1.3.jar:3.1.3]
    at java.base/java.util.LinkedHashMap$LinkedValues.forEach(LinkedHashMap.java:608) ~[na:na]
    at org.springframework.cloud.stream.binding.AbstractBindingLifecycle.start(AbstractBindingLifecycle.java:57) ~[spring-cloud-stream-3.1.3.jar:3.1.3]
    at org.springframework.cloud.stream.binding.InputBindingLifecycle.start(InputBindingLifecycle.java:34) ~[spring-cloud-stream-3.1.3.jar:3.1.3]
    at org.springframework.context.support.DefaultLifecycleProcessor.doStart(DefaultLifecycleProcessor.java:178) ~[spring-context-5.3.8.jar:5.3.8]
    ... 14 common frames omitted

Process finished with exit code 1

我们提供了以下依赖项

implementation 'org.springframework.cloud:spring-cloud-stream'
implementation 'org.springframework.cloud:spring-cloud-stream-binder-kafka'
implementation 'org.springframework.cloud:spring-cloud-stream-binder-rabbit'

rabbitmq

来源：https://stackoverflow.com/questions/71216982/spring-cloud-stream-connect-to-multiple-hosts-for-single-binder-rabbitmq

2条答案

按热度按时间

omhiaaxx1#

在为rabbit1和rabbit2添加活页夹配置后，解决了该问题：
下面是我尝试并成功使用消息的示例配置

spring:
  cloud:
    stream:
      function:
        definition: processFirstConsumer;processSecondConsumer
      bindings:
        processFirstConsumer-in-0:
          group: allocation
          destination: userMessage1
          binder: rabbit1
        processSecondConsumer-in-0:
          group: allocation
          destination: userMessage2
          binder: rabbit2
      binder:
        rabbit1:
          processFirstConsumer-in-0:
            consumer:
              bindingRoutingKey: routing.key.1
          binder:
            nodes: address1
        rabbit2:
          processSecondConsumer-in-0:
            consumer:
              bindingRoutingKey: routing.key.2
          binder:
            nodes: address2
      binders:
        rabbit1:
          type: rabbit
          environment:
            spring:
              rabbitmq:
                host: localhost
                port: 5672
                username: guest
                password: guest
                virtual-host: /
        rabbit2:
          type: rabbit
          environment:
            spring:
              rabbitmq:
                host: localhost
                port: 5672
                username: guest
                password: guest
                virtual-host: /

赞(0）回复(0）举报 2022-11-08

8wtpewkr2#

我费了很大的劲才让这个程序运行起来。尤其是路由键。最后我的解决方案和这里显示的有很大的不同。希望它能在未来帮助到别人：

spring:
  cloud:
    stream:
      binders:
        rabbit1:
          type: rabbit
          environment:
            spring:
              rabbitmq:
                addresses: some.server.com:5672,some.other.server.com:5672
                username: someUserName
                password: someUserPassword
        rabbit2:
          type: rabbit
          environment:
            spring:
              rabbitmq:
                addresses: some.rabbit2.server.us:5672,someother.rabbit2.server.de:5672
                username: secondRabbitUserName
                password: secondRabbitPassword
      function:
        definition: someReceiver;anotherRec
      rabbit:
        bindings:
          someReceiver-in-0:
            consumer:
              auto-bind-dlq: true
              republishToDlq: true
              bindingRoutingKey: some.routing.key.#
          anotherRec-in-0:
            consumer:
              bindingRoutingKey: finished.#
          firstAction-out-0:
            producer:
              routingKeyExpression: "'switch.'+headers.someValue+'.'+headers.someOtherValue"
              bindingRoutingKey: switch.#
          userNotification-out-0:
            producer:
              routingKeyExpression: "'switch.someKeyExpression'"
              bindingRoutingKey: switch.#
      bindings:
        anotherRec-in-0:
          destination: reciving.exchange.name
          group: some-queue-name-v7
          binder: rabbit1
        firstAction-out-0:
          destination: some.exhange.name.1
          binder: rabbit1
        someReceiver-in-0:
          destination: another.exchange.name.1
          group: queueName
          binder: rabbit2
        userNotification-out-0:
          destination: the.exchange.name
          binder: rabbit2

赞(0）回复(0）举报 2022-11-08

我来回答

Spring Cloud Stream可连接到多个主机以实现单个绑定器(RabbitMQ)

2条答案

相关问题

热门标签

最新问答