Spring cloud stream - 为 Rabbitmq 声明目标和组属性

Question

我有生产者和消费者 spring 云流应用程序， 我们不知道我们将首先启动哪个应用程序。它可以是生产者或消费者。

当我们在application.properties中为生产者和消费者应用程序声明destination(Exchange),group(queue)的属性时，如果destination(Exchange),group(queue)已经存在，就会这样尝试使用现有的重新创建，否则创建。

1.如果生产者先启动，则需要创建目的地和组，消费者将使用相同的目的地和组，而无需再次尝试创建。

2.如果消费者先启动，则需要创建目的地和组，生产者将使用相同的目的地和组，而无需再次尝试创建。

需要生产者和消费者 application.properties 来满足上述要求。

制作人：

public interface FoodOrderPublisher {

      String OUTPUT = "foodOrderPublishChannel";

      @Output(OUTPUT)
      MessageChannel create();
}

spring.cloud.stream.bindings.foodOrderPublishChannel.destination=foodOrders
spring.cloud.stream.bindings.foodOrderPublishChannel.group=foodOrdersQueue
spring.cloud.stream.bindings.foodOrderPublishChannel.producer.requiredGroups=foodOrdersQueue
spring.cloud.stream.default.contentType=application/json

消费者：

spring.cloud.stream.bindings.foodOrderRecieveChannel.destination=foodOrders
spring.cloud.stream.bindings.foodOrderRecieveChannel.group=foodOrdersQueue

public interface FoodOrderConsumer {

    String INPUT = "foodOrderRecieveChannel";

    @Input(INPUT)
    SubscribableChannel receive();

}

上面的代码：

生产者应用程序通过创建目的地和组来正确启动，因为消费者给出以下错误：

2018-12-11 20:43:41.607 INFO 1980 --- [main] c.s.b.r.p.RabbitExchangeQueueProvisioner：声明入站队列：foodOrders.foodOrdersQueue，绑定到：foodOrders 2018-12-11 20:43:41.739 WARN 1980 --- [main] os.amqp.rabbit.core.RabbitAdmin：未能声明交换：交换 [name=foodOrders，type=topic，durable=true，autoDelete=false， internal=false, arguments={}], 继续... org.springframework.amqp.AmqpIOException: java.io.IOException 2018-12-11 20:43:41.753 ERROR 1980 --- [127.0.0.1:5672] c.r.c.impl.ForgivingExceptionHandler：发生意外的连接驱动程序错误

java.net.SocketException: 套接字关闭

Answer 1

设置requiredGroups property on the producer side。

必需组

生产者必须确保消息传递到的组的逗号分隔列表，即使它们在创建后开始（例如，通过在 RabbitMQ 中预先创建持久队列）。

.

无需再次尝试创建。

双方都会尝试创建，但由于操作是幂等的，第二方实际上什么都不做。