Kafka Streams-根据Streams数据发送不同的主题

Kafka Streams-根据Streams数据发送不同的主题

您可以使用

branch

方法来拆分流。此方法使用谓词将源流分成几个流。

以下代码取自kafka-streams-examples：

KStream<String, OrderValue>[] forks = ordersWithTotals.branch(    (id, orderValue) -> orderValue.getValue() >= FRAUD_LIMIT,    (id, orderValue) -> orderValue.getValue() < FRAUD_LIMIT);forks[0].mapValues(    orderValue -> new OrderValidation(orderValue.getOrder().getId(), FRAUD_CHECK, FAIL))    .to(ORDER_VALIDATIONS.name(), Produced        .with(ORDER_VALIDATIONS.keySerde(), ORDER_VALIDATIONS.valueSerde()));forks[1].mapValues(    orderValue -> new OrderValidation(orderValue.getOrder().getId(), FRAUD_CHECK, PASS))    .to(ORDER_VALIDATIONS.name(), Produced  .with(ORDER_VALIDATIONS.keySerde(), ORDER_VALIDATIONS.valueSerde()));