搭建完成的zookeeper和kafka集群怎么使用

搭建好Zookeeper和Kafka集群后，你可以按照以下步骤进行基本操作：

1. **配置连接**: 确保客户端可以正确连接到Zookeeper集群，通常是通过IP地址和端口（默认为2181）。Zookeeper负责维护Kafka的元数据信息。

2. **创建主题**: 使用Kafka命令行工具（kafka-topics.sh）或者Kafka API，在Zookeeper上创建一个新的消息主题，指定分区数、复制因子等属性。

bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --topic my-topic --partitions 3 --replication-factor 3

3. **生产者发送消息**: 创建一个生产者实例，设置Zookeeper和Kafka服务器地址，然后开始发送消息到指定的主题。

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("acks", "all");
Producer<String, String> producer = new KafkaProducer<>(props);
producer.send(new ProducerRecord<>("my-topic", "key", "value"));
producer.close();

4. **消费者消费消息**: 创建消费者实例，同样设置Zookeeper和服务器地址，然后订阅主题并接收消息。

Properties consumerProps = new Properties();
consumerProps.put("bootstrap.servers", "localhost:9092");
consumerProps.put("group.id", "test-consumer-group");
Consumer<String, String> consumer = new KafkaConsumer<>(consumerProps);
consumer.subscribe(Arrays.asList("my-topic"));
while (true) {
    ConsumerRecords<String, String> records = consumer.poll(100);
    for (ConsumerRecord<String, String> record : records) {
        System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
    }
}
consumer.close();

**相关问题--:**
1. 如何验证Zookeeper和Kafka之间的通信是否正常？
2. 如果有多个生产者和消费者，如何处理消息的同步和分发？
3. 在大规模部署时，如何管理多个Zookeeper集群和Kafka节点？