Spring Cloud Stream主题分区KStream读/写

Question

我有多个微服务和前面的API，喜欢使用相同的主题为事件每个域事件在单独的分区，我能够配置spring kafka绑定器发送到不同的分区使用

spring.cloud.stream.bindings.<channel>.producer.partition-key- extractor-name= 

实现

我的问题是，我能否将Kstream绑定器配置为只能为@input和@output使用分区。

spring.cloud.stream.kafka.streams.bindings.<channel>.producer.configuration.partitioner.class=

Answer 1

您是否确定地将记录发送到某个分区？换句话说，您知道每个键的实际分区吗？如果只提供PartitionKeyExtractorStrategy，那么绑定器将任意选择一个分区来发送该记录。如果希望使其具有确定性，那么可以在生产者端提供PartitionSelectorClass作为属性（实现接口PartitionSelectorStrategy)。该接口允许您根据键选择分区。假设您希望将所有带有键UUID-1的记录发送到分区1，并且通过PartitionSelectorStrategy实现对其进行编码。这意味着kafka streams处理器知道键为UUID-1的记录来自分区1。有了这些假设，您可以在kafka streams处理器中执行以下操作。这基本上是为你的其他问题提供的这个答案的变体。

@StreamListener("requesti")
@SendTo("responseo")
public KStream<UUID,Account> process(KStream<UUID, Account> events) {


        return  events.transform(() -> new Transformer<UUID, Account, KeyValue<UUID, Account>>() {
            ProcessorContext context;

            @Override
            public void init(ProcessorContext context) {
                this.context = context;
            }

            @Override
            public KeyValue<UUID,Account> transform(UUID key, Account value) {
                if (this.context.partition() == 1) {
                    //your processing logic
                    return KeyValue.pair(key, value);
                }
                return null;
            }

            @Override
            public void close() {

            }
        });
    }

通过上面的代码，基本上可以过滤掉transform方法中所有不相关的分区。仍然存在将出站数据发送到特定分区的问题。如果按照上面的代码执行，那么绑定器将把数据发送到任意分区（这可能是为绑定器添加的一个很好的特性）。但是，在本例中，如果希望出站记录登陆到确定性分区上，则可以直接使用Kafka流。见下文。

@StreamListener("requesti")
public void process(KStream<UUID, Account> events) {


    final KStream<UUID, Account> transformed = events.transform(() -> new Transformer<UUID, Account, KeyValue<UUID, Account>>() {
            ProcessorContext context;


            @Override
            public void init(ProcessorContext context) {
                this.context = context;
            }

            @Override
            public KeyValue<UUID, Account> transform(UUID key, Account value) {
                if (this.context.partition() == 1) {
                    //your processing logic
                    return KeyValue.pair(key, value);
                }
                return null;
            }

            @Override
            public void close() {

            }
        });


        transformed.to("outputTopic", Produced.with(new JsonSerde<>(), new JsonSerde<>(), new CustomStreamPartitioner()));
    }

    class CustomStreamPartitioner implements StreamPartitioner<UUID, Account> {

        @Override
        public Integer partition(String topic, UUID key, Account value, int numPartitions) {
            return 1; //change to the right partition based on the key.
        }
    }