java - 如何使用来自 KStream 输出主题的消息

Question

我正在使用 KStream.to("outputtopic"); 编写输出主题在 apache 文档中提到它将自动创建传递给 to() 的主题。如何使用来自该主题的消息？

我可以使用 consumer.subscribe() 来输出主题和轮询消息吗？

        KStreamBuilder builder = new KStreamBuilder();

        builder.stream(topic).filterNot((k, v) -> {
            v.toString().contains(tid);
        }).to("outputtopic");

        streams = new KafkaStreams(builder, config);
        streams.start();

        consumer.subscribe(Arrays.asList("outputtopic"));

Answer 1

builder.stream(topic).filterNot((k, v) -> {
            v.toString().contains(tid);
        }) // i.e., without the last `to()` method

这一系列方法的结果是一个KStream。如果您的问题是关于如何从同一应用程序中继续操作此结果KStream，那么请执行以下操作:

KStream<..., ...> myStream = builder.stream(topic).filterNot((k, v) -> {
            v.toString().contains(tid);
        });
myStream.to("outputtopic");

// Then continue to use the `myStream` instance for further work.
myStream.map(....).aggregate(...);

如果您的问题是关于如何从不同的应用程序读取输出主题，那么您可以通过从另一个 Kafka Streams 应用程序、KSQL、普通 Kafka 消费者(通过订阅)等