如何动态将同一Kafka事件路由至多个主题？已试用TopicNameExtractor

当TopicNameExtractor只能返回单个主题时，你可以通过以下几种方式实现基于条件的多主题路由：

方法1：使用branch()拆分流后分别输出 #

branch()方法可以根据你定义的多个谓词条件，将原数据流拆分成多个子流，每个子流对应一组符合条件的事件，之后你可以将每个子流单独输出到对应的主题。这种方式适合将事件按互斥条件分发到不同主题。

// 定义拆分条件：按事件类型区分
Predicate<KeyValue<String, Event>> isTypeA = (key, value) -> value.getType().equals("TYPE_A");
Predicate<KeyValue<String, Event>> isTypeB = (key, value) -> value.getType().equals("TYPE_B");
Predicate<KeyValue<String, Event>> isTypeC = (key, value) -> value.getType().equals("TYPE_C");

// 拆分原流为多个子流
KStream<String, Event>[] splitStreams = originalStream.branch(isTypeA, isTypeB, isTypeC);

// 每个子流输出到对应主题
splitStreams[0].to("topic-a");
splitStreams[1].to("topic-b");
splitStreams[2].to("topic-c");

方法2：使用foreach()手动发送到多主题 #

如果同一个事件需要同时发送到多个符合条件的主题（非互斥场景），可以使用foreach()遍历每条消息，根据条件判断目标主题列表，再手动发送消息到这些主题。

// 假设已提前初始化好线程安全的KafkaProducer
KafkaProducer<String, Event> producer = new KafkaProducer<>(producerConfigs);

originalStream.foreach((key, value) -> {
    List<String> targetTopics = new ArrayList<>();
    
    // 根据事件属性判断目标主题
    if (value.isHighPriority()) {
        targetTopics.add("high-priority-topic");
    }
    if (value.needsAudit()) {
        targetTopics.add("audit-log-topic");
    }
    if (value.isForAnalytics()) {
        targetTopics.add("analytics-topic");
    }
    
    // 批量发送到多个主题
    for (String topic : targetTopics) {
        producer.send(new ProducerRecord<>(topic, key, value));
    }
});

注意：确保KafkaProducer是线程安全的，或通过Kafka Streams的ProcessorContext获取生产者，避免线程安全问题。

方法3：自定义Processor实现复杂路由 #

如果你的路由逻辑比较复杂（比如需要动态生成主题名、多条件组合判断），可以自定义Processor，通过ProcessorContext将消息转发到多个主题。

自定义Processor类 #

public class MultiTopicRouter implements Processor<String, Event> {
    private ProcessorContext context;

    @Override
    public void init(ProcessorContext context) {
        this.context = context;
    }

    @Override
    public void process(String key, Event value) {
        // 根据事件属性确定目标主题列表
        List<String> targetTopics = getTargetTopics(value);
        
        // 转发消息到每个目标主题
        for (String topic : targetTopics) {
            context.forward(key, value, To.child(topic));
        }
    }

    private List<String> getTargetTopics(Event value) {
        List<String> topics = new ArrayList<>();
        if (value.getCategory().equals("PAYMENT")) {
            topics.add("payments-processing");
            topics.add("financial-audit");
        } else if (value.getCategory().equals("USER")) {
            topics.add("user-activity");
            topics.add("user-profile-updates");
        }
        return topics;
    }

    @Override
    public void close() {}
}

在拓扑中使用自定义Processor #

Topology topology = new Topology();
topology.addSource("INPUT_SOURCE", "input-topic")
        .addProcessor("MULTI_ROUTER", MultiTopicRouter::new, "INPUT_SOURCE")
        // 为每个目标主题添加Sink
        .addSink("SINK_PAYMENTS", "payments-processing", "MULTI_ROUTER")
        .addSink("SINK_AUDIT", "financial-audit", "MULTI_ROUTER")
        .addSink("SINK_USER_ACTIVITY", "user-activity", "MULTI_ROUTER")
        .addSink("SINK_USER_PROFILE", "user-profile-updates", "MULTI_ROUTER");

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