Kafka通过多个分区和多个消费者线程来增加吞吐量

Question

我正在使用kafka流进行某些应用程序。

流流如下

kafkaProducer---->StreamerConsumer1->finalCosumer

我有生产者非常快速地写入数据，我的StreamConsumer将使用某个进程映射每个流并将流转发到其他主题。

在我的StreamCosumer地图中，我添加了自己的mapper函数，它实际上试图保存其相关数据，如下所示

public void checkRecord(T1 key, T2 value) {
 switch(T1.toString()){
 case "key1":
  //Get relavant fileds from value and perisit in db 
   break;
   case "key2":
     //Get relavant fileds from value and perisit in db 
   break;
 }
}


KStream<String, KafkaStatusRecordWrapper> pDStream[] = myStream.map(this::checkRecord).branch((key, value)-> value.isSuccess(),(key, value)-> !value.isSuccess());

pDStream[0].mapValues(value -> transformer(value)).to("other_topic",Produced.with(stringSerde, stringSerde));   

现在我的checkRecord记录消费者功能是单线程的，几乎需要300ms（由于一些业务逻辑和数据库持久存在，我无法避免）返回。

我不能增加分区的数量，因为我们的infra有一些限制，也是由于以下限制

More Partitions Requires More Open File Handles
More Partitions May Increase Unavailability
More Partitions May Increase End-to-end Latency

所以我打算编写多线程流消费者。

但我担心以下几点。

1. 我只需要处理一次记录
2. 切换到另一个线程将导致偏移管理问题。

那么如何提高吞吐量呢？

我的消费者有足够的资源，只使用了40％的资源。

Answer 1

您可以设置流配置num.stream.threads以配置线程数。最大值可以是最大分区数。它有助于增加应用程序实例的并行性。

假设您的主题有4个分区，您可以设置以下内容：

properties.set("num.stream.threads",4);