此版本仍在开发中，尚未被视为稳定版本。对于最新的稳定版本，请使用 spring-cloud-stream 4.1.4！spring-doc.cn

使用 RabbitMQ Binder 进行分区

RabbitMQ 本身不支持分区。spring-doc.cn

有时，将数据发送到特定分区是有利的 — 例如，当您想要严格排序消息处理时，特定客户的所有消息都应发送到同一分区。spring-doc.cn

通过将每个分区的队列绑定到目标 Exchange 来提供分区。RabbitMessageChannelBinderspring-doc.cn

以下 Java 和 YAML 示例展示了如何配置 producer：spring-doc.cn

制作人

@SpringBootApplication
public class RabbitPartitionProducerApplication {

    private static final Random RANDOM = new Random(System.currentTimeMillis());

    private static final String[] data = new String[] {
            "abc1", "def1", "qux1",
            "abc2", "def2", "qux2",
            "abc3", "def3", "qux3",
            "abc4", "def4", "qux4",
            };

    public static void main(String[] args) {
        new SpringApplicationBuilder(RabbitPartitionProducerApplication.class)
            .web(false)
            .run(args);
    }

    @Bean
    public Supplier<Message<?>> generate() {
        return () -> {
            String value = data[RANDOM.nextInt(data.length)];
            System.out.println("Sending: " + value);
            return MessageBuilder.withPayload(value)
                    .setHeader("partitionKey", value)
                    .build();
        };
    }

}

application.yml

    spring:
      cloud:
        stream:
          bindings:
            generate-out-0:
              destination: partitioned.destination
              producer:
                partitioned: true
                partition-key-expression: headers['partitionKey']
                partition-count: 2
                required-groups:
                - myGroup

前面示例中的配置使用默认的分区（）。这可能会也可能不会提供适当平衡的算法，具体取决于键值。您可以使用 or 属性覆盖此默认值。key.hashCode() % partitionCountpartitionSelectorExpressionpartitionSelectorClassspring-doc.cn

仅当您需要在部署创建者时预置使用者队列时，该属性才是必需的。否则，发送到分区的任何消息都将丢失，直到部署相应的使用者为止。required-groupsspring-doc.cn

以下配置预置主题交换：spring-doc.cn

以下队列绑定到该 Exchange：spring-doc.cn

以下绑定将队列关联到交换：spring-doc.cn

以下 Java 和 YAML 示例延续了前面的示例，并展示了如何配置使用者：spring-doc.cn

消费者

@SpringBootApplication
public class RabbitPartitionConsumerApplication {

    public static void main(String[] args) {
        new SpringApplicationBuilder(RabbitPartitionConsumerApplication.class)
            .web(false)
            .run(args);
    }

    @Bean
    public Consumer<Message<String>> listen() {
        return message -> {
            String queue =- message.getHeaders().get(AmqpHeaders.CONSUMER_QUEUE);
            System.out.println(in + " received from queue " + queue);
        };
    }

}

application.yml

    spring:
      cloud:
        stream:
          bindings:
            listen-in-0:
              destination: partitioned.destination
              group: myGroup
              consumer:
                partitioned: true
                instance-index: 0

不支持动态扩展。每个分区必须至少有一个使用者。 consumer's 用于指示使用的分区。 Cloud Foundry 等平台只能有一个实例具有 .RabbitMessageChannelBinderinstanceIndexinstanceIndex