Kafka消费者处理不同主题的同步消息。

下面是一个使用Java语言的示例代码，展示了如何使用Kafka消费者处理不同主题的同步消息：

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

public class KafkaConsumerExample {
    public static void main(String[] args) {
        // 设置Kafka消费者的配置
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); // Kafka服务器地址和端口
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-consumer-group"); // 消费者组ID
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); // 键的反序列化器
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); // 值的反序列化器

        // 创建Kafka消费者
        Consumer<String, String> consumer = new KafkaConsumer<>(props);

        // 订阅多个主题
        consumer.subscribe(Arrays.asList("topic1", "topic2"));

        // 持续拉取和处理消息
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));

            // 处理每条消息
            for (ConsumerRecord<String, String> record : records) {
                String topic = record.topic();
                String key = record.key();
                String value = record.value();

                // 处理消息的逻辑
                System.out.println("Topic: " + topic + ", Key: " + key + ", Value: " + value);
            }
        }
    }
}

上述代码创建了一个Kafka消费者，并通过props对象设置了消费者的配置信息。然后，使用subscribe方法订阅了两个主题（"topic1"和"topic2"）。之后，在一个无限循环中，通过poll方法拉取消息，然后使用for循环遍历每条消息进行处理。

在处理消息的逻辑中，可以根据需要自定义具体的处理方式。上述示例代码只是简单地将消息的主题、键和值打印出来，实际使用时可以根据业务需求进行相应的处理。