Spring云流没有向RabbitMQ发送消息，而是在同一个pod中使用。生产者和消费者在同一个应用程序中

Question

我使用的是Spring Cloud Stream和RabbitMQ，我在同一个应用程序中配置了生产者和消费者，但使用了两个不同的通道。问题是，我的应用程序有五个副本在Kubernetes上运行，但只有发送消息的pod会使用它们，其他四个不会使用。在RabbitMQ控制台上，没有收到任何消息，消息速率图不会更改。当我注释OrdersListener时，消息会成功发送到RabbitMQ集群。有人知道可能出了什么问题吗？

以下是制片人频道：

@Component
public interface OrdersChannel {
   @Output("Orders")
   MessageChannel publishOrders();
}

这是消费者渠道：

@Component
public interface OrdersListenerChannel {
   String BINDING = "Orders";

   @Input(BINDING)
   SubscribableChannel input();
}

以下是生成消息的服务：

@Service
@RequiredArgsConstructor
@EnableBinding(OrdersChannel.class)
public class OrdersEventService {

    private final OrdersChannel ordersChannel;

    public void sendOrderEvents(final Set<Order> orders) {
        orders.parallelStream().forEach(order ->
            ordersChannel.publishOrders()
            .send(message(OrderMessageDTOBuilder.build(order))));
    }

    private static <T> Message<T> message(T val) {
        return MessageBuilder.withPayload(val).build();
    }
}

下面是听众：

@Component
@RequiredArgsConstructor
@EnableBinding(OrdersListenerChannel.class)
public class OrdersListener {

    private final OrderService orderService;

    @StreamListener(OrdersListenerChannel.BINDING)
    public void listen(final OrderMessageDTO orderMessageDTO) {
       // orderService method call
    }
}

我的application.properties是这样的：

spring.cloud.stream.bindings.Orders.destination=GeneralExchange
spring.cloud.stream.bindings.Orders.producer.requiredGroups=OrdersQueue
spring.cloud.stream.rabbit.bindings.Orders.producer.bindingRoutingKey=OrdersQueue
spring.cloud.stream.rabbit.bindings.Orders.producer.routingKeyExpression='OrdersQueue'


spring.cloud.stream.bindings.OrdersListenerChannel.group=OrdersQueue
spring.cloud.stream.bindings.OrdersListenerChannel.consumer.max-attempts=3
spring.cloud.stream.bindings.OrdersListenerChannel.consumer.concurrency=10
spring.cloud.stream.rabbit.bindings.OrdersListenerChannel.consumer.bind-queue=false
spring.cloud.stream.rabbit.bindings.OrdersListenerChannel.consumer.republish-to-dlq=false

Answer 1

所以，您要问几个问题。首先，您的配置和编程模型（基于注释）早已弃用，实际上将在下一个主要版本中删除。我们渴望切换到更简单的函数式编程模型。

关于你的副本。我认为这与你所观察到的没有任何关系。您实际上是在说您有五个消费者，但只有一个收到消息。这意味着您没有在RabbiotMQ中配置发布子模型。这类似于未配置或未理解交换类型和队列路由密钥之间的RabbitMQ关系。您的exchange必须是正确路由到绑定队列的类型（例如，作为路由密钥）