添加空处理器时 Kafka Streams 处理速度变慢

Question

考虑这个 Kafka Streams 驱动程序

public class TestDriver {

    private static final String SOURCE = "SOURCE";

    public static void main(String[] args) throws Exception {

        ProtoDeserializer<Message> protoDeserializer = new ProtoDeserializer<>(Message.parser());
        ProtoSerializer<Message> protoSerializer = new ProtoSerializer<>();

        StringDeserializer stringDerializer = new StringDeserializer();
        StringSerializer stringSerializer = new StringSerializer();

        Topology topologyBuilder = new Topology();
        topologyBuilder.addSource(SOURCE, stringDerializer, protoDeserializer, "input-messages")

            .addProcessor(DummyProcessor.NAME, DummyProcessor::new, SOURCE)

            .addSink("MAIN", "output-messages", stringSerializer, protoSerializer, DummyProcessor.NAME)
        ;

        KafkaStreams streams = new KafkaStreams(topologyBuilder, getConfig());
        streams.cleanUp();
        streams.start();

        System.out.println(streams.toString());

        Runtime.getRuntime().addShutdownHook(new Thread(streams::close));

    }

    private static Properties getConfig() {
        Properties config = new Properties();
        config.put(StreamsConfig.CLIENT_ID_CONFIG, "test.stream-processor");
        config.put(StreamsConfig.APPLICATION_ID_CONFIG, "test.stream-processor");
        config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "broker-1:9092,broker-2:9092,broker-3:9092");
        config.put(StreamsConfig.REPLICATION_FACTOR_CONFIG, 3);
        config.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, 10);
        config.put(StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, WallclockTimestampExtractor.class);
        return config;
    }
}

问题是在topology中没有加processor的情况下（不包含.addProcessor()），从source到sink的处理速度还可以（意思是我目前生产25kmessages/s和赶上来没问题）。

但是，当添加DummyProcessor时，它突然处理3k messages/s max（600k字节）。

DummyProcessor 基本上什么都不做：

public class DummyProcessor extends AbstractProcessor<String, Message> {

    public static final String NAME = "DUMMY_PROCESSOR";

    public void process(String key, Message originalMessage) {
        context().forward(key, originalMessage);
        context().commit();
    }
}

添加单个 "empty" 处理器是否会增加 Streams 性能的开销？是什么原因造成的？ Kafka Streams 是否如此智能以至于当没有处理器时它不执行 protobuf serde 并且只转发接收到的数据？无论如何要加快速度？

以这样的速度，我需要 x 数千个 cpu 可用线程来处理我的所有数据，因为 25k messages/s 是我拥有数据的 1% .听起来很多。

Answer 1

问题是由于经常请求提交造成的。

您根本不需要调用 ProcessorContext:commit()。基于 commit.interval.ms 属性的 Kafka Streams 执行提交（默认：30000 ms）。如果恰好一次语义被设置为不同的值。您可以在 https://kafka.apache.org/documentation/#streamsconfigs.

中查看详细信息

如果在某些用例中您需要更频繁地提交，您可以调用 ProcessorContext:commit()。但是您必须记住，提交不是立即（直接）进行的。它只设置标志以尽快提交。

添加空处理器时 Kafka Streams 处理速度变慢

Kafka Streams processing speed slows when empty processor is added

java

apache-kafka

apache-kafka-streams