Kafka事件路由器：过滤和基于内容的路由，如何实现？

Kafka提供了许多API来实现不同类型的路由行为，包括过滤和基于内容的路由。以下是基于Java语言的示例代码：

1. 过滤路由

在这种路由中，只有满足特定条件的事件才会被路由到特定的目标主题。在下面的示例中，我们过滤出操作系统为Windows的事件，并将它们路由到名为win_events的主题中。

//创建Kafka处理程序
KafkaProducer<String, String> producer = new KafkaProducer<>(props);

//创建Consumer处理程序
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

//定义过滤条件
Pattern pattern = Pattern.compile("windows");

// 路由事件
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        if (pattern.matcher(record.value()).find()) {
            producer.send(new ProducerRecord<String, String>("win_events", record.value()));
        }
    }
}

2. 基于内容的路由

在这种路由中，事件将根据其内容路由到不同的主题。在下面的示例中，我们路由包含“error”字眼的事件到名为error_events的主题中，路由包含“warning”字眼的时间到名为warning_events的主题中。

//定义路由规则
HashMap<String, String> topicMap = new HashMap<String, String>();
topicMap.put("error", "error_events");
topicMap.put("warning", "warning_events");
topicMap.put("default", "default_events");

// 路由事件
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        for (String key : topicMap.keySet()) {
            if (record.value().contains(key)) {