如何使用 RMQ 和 Spring Cloud 流创建基于分区的生产者? [英] How to use RMQ and spring cloud stream for creating partition based producer?

问题描述

我正在尝试找到 Spring Cloud 流的示例，它在其中为 RMQ 创建基于分区的生产者.我想看看它将如何为这些队列创建绑定，因为 RMQ 本身不支持主题的分区，但它会创建与分区数相等的队列数(我读过这个，我可能错了).首先，我想了解如何在基于分区的生产者的 RMQ 上使用 Spring Cloud 流创建生产者.

I am trying to find example of spring cloud stream where it creates parition based producer for RMQ. I want to see how it will create binding for those queues as RMQ not natively support paritions for topic but it will create number of queues equivalent to number of partitions (I read this,I may be wrong). First I want to understand how to create producer using spring cloud stream on RMQ for partition based producer.

推荐答案

@SpringBootApplication
@EnableBinding(Source.class)
public class So43614477Application implements CommandLineRunner {

    public static void main(String[] args) {
        SpringApplication.run(So43614477Application.class, args);
    }

    @Autowired
    private MessageChannel output;

    @Override
    public void run(String... args) throws Exception {
        output.send(MessageBuilder.withPayload("fiz").setHeader("whichPart", 0).build());
        output.send(MessageBuilder.withPayload("buz").setHeader("whichPart", 1).build());
    }

}

有属性...

spring.cloud.stream.bindings.output.destination=foo
spring.cloud.stream.bindings.output.producer.partition-key-expression=headers['whichPart']
spring.cloud.stream.bindings.output.producer.partition-count=2

spring.cloud.stream.bindings.output.producer.required-groups=bar

我添加了 required-groups 以便您可以看到消费者队列是如何绑定的.

I added the required-groups so you can see how the consumer queues are bound.

foo 交换绑定:

表达式可以是针对消息计算的任何有效表达式(例如 payload.hashCode())；然后根据分区计数修改 % 以确定最终分区.

The expression can be any valid expression evaluated against the message (e.g. payload.hashCode()); it is then modded % against the partition count to determine the final partition.

这篇关于如何使用 RMQ 和 Spring Cloud 流创建基于分区的生产者?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！