要演示Apache Kafka消费者组跨集群的示例,您需要进行以下步骤:
确保您已经安装了Apache Kafka并设置了多个Kafka集群。每个集群应具有不同的broker端口和Zookeeper连接字符串。
在每个Kafka集群中创建一个或多个主题,以便在不同集群之间进行消息传递。
创建一个消费者组,该组将跨多个集群进行消费。这可以通过在消费者的配置中设置group.id属性来实现。
在消费者代码中,使用消费者组ID来创建KafkaConsumer对象,并将适当的Kafka集群的broker列表传递给bootstrap.servers属性。
下面是一个简单的Java代码示例,演示了如何在两个Kafka集群之间进行消息传递:
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 java.time.Duration;
import java.util.Collections;
import java.util.Properties;
public class ConsumerExample {
public static void main(String[] args) {
Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); // 指定第一个Kafka集群的broker列表
props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-consumer-group"); // 设置消费者组ID
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
KafkaConsumer consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("my-topic")); // 订阅主题
while (true) {
ConsumerRecords records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord record : records) {
System.out.printf("Consumed message: topic = %s, partition = %s, offset = %d, key = %s, value = %s%n",
record.topic(), record.partition(), record.offset(), record.key(), record.value());
}
}
}
}
要在第二个Kafka集群上使用相同的消费者组进行消费,请将上述代码复制到另一个Java类中,并修改以下属性:
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9093"); // 指定第二个Kafka集群的broker列表
请确保在每个集群中使用不同的broker列表,并使用相同的消费者组ID进行消费。这样,您的消费者组将跨多个Kafka集群进行消息传递。