1. 安装Zookeeper：
- Kafka依赖于Zookeeper来进行集群协调。确保安装了Zookeeper，并配置好。
2.下载和解压Kafka：
- 从官方网站下载Kafka，并解压到所需的目录。
3.配置Kafka：
- 修改Kafka的配置文件以适应集群环境。主要修改server.properties文件，设置broker.id、listeners、zookeeper.connect等参数。
4.启动Kafka服务器：
- 分别在每个节点上启动Kafka服务器。
点击查看代码
bin/kafka-server-start.sh config/server.properties
5.创建Topic：
- 创建一个或多个Topic，可以在其中的任一节点上执行。
点击查看代码
bin/kafka-topics.sh --create --topic your-topic --partitions 3 --replication-factor 2 --bootstrap-server localhost:9092
6.验证集群状态：
- 可以使用以下命令检查集群状态：
点击查看代码
bin/kafka-topics.sh --list --bootstrap-server localhost:9092
bin/kafka-topics.sh --describe --topic your-topic --bootstrap-server localhost:9092
注意事项：
1.Broker ID唯一性：
- 每个Kafka节点的broker.id必须是唯一的，且集群中的每个节点都应该有一个唯一的标识。
2.Zookeeper连接配置：
- 确保Kafka配置文件中的zookeeper.connect参数正确指向Zookeeper集群。
3.Topic分区和副本配置：
- 根据需求设置Topic的分区数和副本数。副本数可以保证数据的冗余和可用性。
4.网络配置：
- 确保Kafka节点之间的网络能够正常通信，特别是防火墙设置。
5.版本一致性：
- 确保Kafka集群的各个节点使用相同版本的Kafka。
示例：
生产者示例：
点击查看代码
import org.apache.kafka.clients.producer.*;
import java.util.Properties;
public class KafkaProducerExample {
public static void main(String[] args) {
Properties props = new Properties();
props.put("bootstrap.servers", "node1:9092,node2:9092,node3:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
Producer<String, String> producer = new KafkaProducer<>(props);
String topic = "your-topic";
String key = "key1";
String message = "Hello, Kafka!";
ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, message);
producer.send(record, new Callback() {
@Override
public void onCompletion(RecordMetadata metadata, Exception exception) {
if (exception == null) {
System.out.println("Message sent to partition " + metadata.partition() + " with offset " + metadata.offset());
} else {
exception.printStackTrace();
}
}
});
producer.close();
}
}
消费者示例：
点击查看代码
import org.apache.kafka.clients.consumer.*;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
public class KafkaConsumerExample {
public static void main(String[] args) {
Properties props = new Properties();
props.put("bootstrap.servers", "node1:9092,node2:9092,node3:9092");
props.put("group.id", "your-consumer-group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
Consumer<String, String> consumer = new KafkaConsumer<>(props);
String topic = "your-topic";
consumer.subscribe(Collections.singletonList(topic));
while (true) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<String, String> record : records) {
System.out.printf("Consumed record with key %s and value %s%n", record.key(), record.value());
}
}
}
}
标签：示例,--,注意事项,9092,Kafka,topic,props,kafka From： https://www.cnblogs.com/arek/p/17897664.html以上示例中，生产者将消息发送到指定的Topic，而消费者则从同一Topic中消费消息。请注意替换示例代码中的占位符（如your-topic、node1:9092,node2:9092,node3:9092等）为你实际的配置信息。