使用Kafka流对具有相同键值的Kafka消息进行计数

Question

我有一个Java Spring Kafka应用程序，它将Book类型的对象发送到Kafka主题。然后我试着用Kafka流来映射每一条信息，把这本书的作者作为它的关键字。然后，我尝试将它们添加到一个KTable中，该表保存了密钥和拥有该密钥的消息的数量。然后，该表被发送到一个输出kafka主题。

书籍型号：

@Data
public class Book {
    private UUID id;
    private String name;
    private String author;
    private LocalTime date;
}

流结构：

StreamsBuilder streamsBuilder = new StreamsBuilder();
        KStream<String,Book> stream = streamsBuilder.stream("input_topic", Consumed.with(Serdes.String(), new BookSerde()));
        stream = stream.selectKey((key,value)->value.getAuthor());
 
        KTable<String ,Long> keyWithCount = stream.groupBy((key, value) -> value.getAuthor()).count();
        keyWithCount.toStream().to("output_topic", Produced.with(Serdes.String(), Serdes.Long()));
 
        KafkaStreams kafkaStreams = new KafkaStreams(streamsBuilder.build(), config);
        kafkaStreams.start();

运行应用后，输出显示每个唯一键，但随机表情，而不是计数。

在搞砸.peek（）之后，我发现了一件奇怪的事情。在发送到主题之前，我打印了密钥和值，它表明它以应有的方式工作！然而，问题在于，该主题的kafka控制台消费者不会在计数中显示数字，而是显示一些随机符号，例如钻石，心形，笑脸等。

一个例子是，在将作者为“a”的书发送到我的制作人的父主题后，我的流媒体应用程序将其视为第一本书，打印“A 1 ”,但在kafka控制台消费者中，它看起来像“☺".”

Answer 1

我用peek（）的发现应该是赠品。问题不在于代码，而在于 kafka 控制台消费者的配置。默认值反序列化程序无法反序列化长整型值。为了解决这个问题，我不得不在初始化消费者时添加 --value-反序列化程序 “组织.apache.kafka.common.serialization.LongDeserializer”。