您列出的所有方法都是可能的。哪一个是最好的实际上取决于您想要对流程的控制，或者它是一次性操作还是您想要继续运行的东西。
kafka streams提供了一种通过dsl将一个主题转换为另一个主题的简单方法
您可以这样做（演示代码显然不是用于生产的！）：

Properties props = new Properties();
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-wordcount");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

final Serde<byte[]> bytesSerdes = Serdes.ByteArray();
final StreamsBuilder builder = new StreamsBuilder();
KStream<byte[], byte[]> input = builder.stream(
        "input-topic",
        Consumed.with(bytesSerdes, bytesSerdes)
);
input.to("output-topic", Produced.with(bytesSerdes, bytesSerdes));

final KafkaStreams streams = new KafkaStreams(builder.build(), props);
try {
    streams.start();
    Thread.sleep(60000L);
} catch (Exception e) {
    e.printStackTrace();
} finally {
    streams.close();
}