如何向KStream<String, String>中合并指定字符串？

我来帮你解决这个问题！KStream是Kafka Streams里的流式处理抽象，它不像普通集合那样支持直接添加元素，得通过框架提供的API把"cat"注入到数据流的处理链路里。下面给你两种实用的方案：

方案一：在原有流的每个元素后附加"cat" #

如果你的需求是让原流的每一条消息都额外带上"cat"这个字符串，可以用flatMapValues方法，它能把单个输入值转换成多个输出值：

import java.util.ArrayList;
import java.util.List;
import java.util.Arrays;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.Produced;

static void createWordCountStream(final StreamsBuilder builder) {
    final KStream<String, String> input = builder.stream(INPUT_TOPIC);
    
    // 用flatMapValues将原消息和"cat"一起输出
    KStream<String, String> inputWithCat = input.flatMapValues(value -> {
        List<String> combinedValues = new ArrayList<>();
        combinedValues.add(value); // 保留原流的元素
        combinedValues.add("cat"); // 添加目标字符串
        return combinedValues;
    });
    
    // 后续可以继续处理inputWithCat，比如做词频统计
    inputWithCat.flatMapValues(text -> Arrays.asList(text.toLowerCase().split("\\W+")))
                .groupBy((key, word) -> word)
                .count()
                .toStream()
                .to(OUTPUT_TOPIC, Produced.with(Serdes.String(), Serdes.Long()));
}

方案二：独立生成包含"cat"的流并合并 #

如果你需要不管原流有没有数据，都要把"cat"作为一条独立消息加入到流中，可以先创建一个只包含"cat"的流，再用merge方法和原流合并（Kafka Streams 2.5+版本支持fromIterable方法，非常方便）：

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.Consumed;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KeyValue;
import org.apache.kafka.streams.kstream.Produced;
import java.util.Collections;
import java.util.Arrays;

static void createWordCountStream(final StreamsBuilder builder) {
    final KStream<String, String> input = builder.stream(INPUT_TOPIC);
    
    // 创建只包含"cat"的流，key可以自定义，比如用"cat-key"
    KStream<String, String> catStream = KStream.fromIterable(
        Collections.singleton(new KeyValue<>("cat-key", "cat")),
        Consumed.with(Serdes.String(), Serdes.String())
    );
    
    // 合并原流和cat流
    KStream<String, String> mergedStream = input.merge(catStream);
    
    // 后续处理合并后的流
    mergedStream.flatMapValues(text -> Arrays.asList(text.toLowerCase().split("\\W+")))
                .groupBy((key, word) -> word)
                .count()
                .toStream()
                .to(OUTPUT_TOPIC, Produced.with(Serdes.String(), Serdes.Long()));
}

关键说明 #

你之前直接定义String word = "cat"是无效的，因为KStream是流式处理的管道，只有把元素注入到这个管道的处理链路中，才能被后续逻辑识别和处理。如果使用的是低于2.5版本的Kafka Streams，可以用Transformer接口来生成包含"cat"的流，不过写法会繁琐一些，优先推荐升级版本或使用方案一。

内容的提问来源于stack exchange，提问作者Zombie88