弹簧云数据流量缩放与Apache Kafka

Question

我在缩放上遵循了许多春季教程（在这里），但仍未找到使所有消费者忙碌的秘密敲门声。目的是让5个（示例）下沉过程不断忙碌。文档意味着对于简单的源流|处理器|接收器，我将使用以下流属性：

deployer.sink.count=5
app.processor.producer.partitionKeyExpression=headers['X-PartitionKey']?:''
app.sink.spring.cloud.stream.kafka.binder.autoAddPartitions=true
app.sink.spring.cloud.stream.kafka.binder.minPartitionCount=5

自然，处理器添加了一个名为X-PartitionKey的标头字段，最终完全具有唯一的唯一，以至于应该足够平衡。

我发现只有2或3个接收到消息，其余的坐着。感觉就像前几个观看多个分区一样，其他人只是坐在待机中说没有发现分区的承诺偏移。我可以通过使用kowl看到消息获得了唯一的分区，但是负载永远不会平衡以反射。

我在某个地方缺少配置吗？这是Kafka活页夹问题吗？

更新＃1

我注意到每个实例都没有获得唯一的客户端。不确定这是否相关。

Instance 1 - [Consumer clientId=consumer-mystream-2, groupId=mystream]
Instance 2 - [Consumer clientId=consumer-mystream-3, groupId=mystream]
Instance 3 - [Consumer clientId=consumer-mystream-3, groupId=mystream]
Instance 4 - [Consumer clientId=consumer-mystream-3, groupId=mystream]
Instance 5 - [Consumer clientId=consumer-mystream-2, groupId=mystream]

更新＃2

使用以下3个属性，所有5个实例都很忙。此方法似乎绕过SCST并使用MessageKey而不是PartitionKey。还不是完美的，但是更好。

deployer.sink.count=5
app.processor.spring.cloud.stream.kafka.binder.messageKeyExpression=headers['X-PartitionKey']?:''
app.sink.spring.cloud.stream.kafka.binder.autoAddPartitions=true

更新＃3

添加spring.cloud.stream.bindings.output.producer.partition-count-count等于Kafka分区的实际数量似乎已经解决了问题。如果您的Deployer。*。计数属性少于分区的实际数量，则将不会分配多余的分区，但是无论如何，消费者将被分配给他们，并且可能会安置。

Answer 1

您可以清楚地看到在这5个实例中均匀分配的10个分区，例如

partitions assigned: [mystream.processor-7, mystream.processor-6]

1: 3,2
2: 0,1
3: 9,8
4: 5,4
5: 7,6

（在所有5个实例中）的消息

Found no committed offset for partition mystream.processor-8

仅表示该群体ID从未为该分区承担偏移。

这些条件对于这种情况是正常的：

2022-05-04 17：42：444.067信息[接收器，] 1 --- [container-0-c-1] oakccinternals.consumerCoordinator：[commuter clientId = compliter clientid = consumer-mystream-2，groupid = mystream = mystream = mystream = mystream = mystream]没有发现分区的偏移偏移。

2022-05-04 17：42：444.067信息[接收器，] 1 --- [container-0-c-1] oakccinternals.consumerCoordinator：[commuter clientId = compliter clientid = consumer-mystream-2，groupid = mystream = mystream = mystream = mystream = mystream]发现没有承诺的分区缩写mystream.processor-6

2022-05-04 17：42：44.163信息[sink，] 1 --- [Container-0-C-1] oakccinternals.subscriptionState：[commuter clientId = compliter clientId = consumer-mystream-2，groupid = mystream = mystream = mystream]重置分区mystream.processor-7的偏移量{offset = 0，offsetepoch = optional.empty.empty，currentLeader = definess = drademandePoch {strafe = dataflow-kafka-0.dataflow-kafka-0.dataflow-kafka-headless.defelest.default.default.svc.cluster.svc.cluster.svc.cluster.svc.cluster.svc.cluster.svc.cluster.svc.cluster.svc.cluster.svc.cluster.local ：9092（ID：0机架：null）]，epoch = 12}}。

2022-05-04 17：42：444.167信息[sink，] 1 --- [container-0-c-1] oakccinternals.subscriptionState：[commuter clientId = compliter clientid = consumer-mystream-2，groupid = mystream = mystream = mystream]重置分区的偏移mystream.processor-6以定位fetchPosition {offset = 0，offsetEpoch = optional.empty.empty，currentLeader = dradingAndePoch {dr量图= dataflow-kafka-1.dataflow-kafka-1.dataflow-kafka-headless.default.default.svc.cluster.svc.cluster.svc.cluster.svc.cluster.svc.cluster.svc.cluster.local ：9092（ID：1架子：null）]，epoch = 10}}。

主要消费者是消费者 - 潮汐-2；重复的日志进一步向下是consumer clientId = consumer -mystream -3 - 我认为这是用于健康检查的消费者 - 它具有相同的组，但从未订阅过这个主题，也许是因为这个聊天是因为在同一组中。

一旦主要消费者承担纪录的偏移，chat不休就消失了。

无论如何，如果您将记录发送给10个分区中的每个分区，每个实例肯定会得到2个。

编辑

它可以根据我的预期使用本机或粘合剂分区。 partitionCount会自动调整以匹配分区的实际数量，并且在计算分区时用作模量。

@SpringBootApplication
public class So72115288Application {

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

    List<AtomicInteger> receivedPartitions = new ArrayList<>();

    CountDownLatch latch = new CountDownLatch(100);

    @Bean
    public Consumer<Message<String>> input() {
        return msg -> {
            this.receivedPartitions.get(msg.getHeaders()
                    .get(KafkaHeaders.RECEIVED_PARTITION_ID, Integer.class))
            .incrementAndGet();
            this.latch.countDown();
        };
    }

    @Bean
    ApplicationRunner runner(StreamBridge bridge) {
        for (int i = 0; i < 10; i++) {
            this.receivedPartitions.add(new AtomicInteger());
        }
        return args -> {
            // native partitioning
//          IntStream.range(0, 100).forEach(i -> bridge.send("output", new GenericMessage<>("foo",
//                  Collections.singletonMap(KafkaHeaders.MESSAGE_KEY, ("key" + i).getBytes()))));
            // binder partitioning
            IntStream.range(0, 100).forEach(i -> bridge.send("output", new GenericMessage<>("foo",
                    Collections.singletonMap("custom", ("key" + i).getBytes()))));
            this.latch.await(60, TimeUnit.SECONDS);
            System.out.println(this.receivedPartitions.stream()
                    .map(ai -> ai.get())
                    .collect(Collectors.toList()));
        };
    }

}

spring.cloud.stream.bindings.input-in-0.group=foo
spring.cloud.stream.bindings.output.producer.partition-count=5
spring.cloud.stream.bindings.output.producer.partition-key-expression=headers['custom']
spring.cloud.stream.bindings.output.destination=input-in-0

spring.cloud.stream.kafka.binder.min-partition-count=5
spring.cloud.stream.kafka.binder.auto-add-partitions=true

[10, 13, 8, 10, 13, 6, 9, 16, 12, 3]