一、启动集群

对于只有一个broker的kafka来说，这就是一个节点数为1的集群。现在，让我们启动一个几点数为3的集群。

1、复制出另外两份kafka配置文件

cp config/server.properties config/server-1.properties
cp config/server.properties config/server-2.properties

# 修改server-1.properties中的以下几项配置为
broker.id=1
listeners=PLAINTEXT://:9093
log.dirs=/tmp/kafka-logs-1

# 修改server-2.properties中的以下几项配置为
broker.id=2
listeners=PLAINTEXT://:9094
log.dirs=/tmp/kafka-logs-2

broker.id是节点在kafka集群中的唯一标识名称，除此之外，我们必须重写端口和日志目录。

2、依次启动zookeeper和三个kafka节点

# 启动zookeeper
bin/zookeeper-server-start.sh config/zookeeper.properties

# 启动三个kafka节点，并后台运行
bin/kafka-server-start.sh config/server.properties &

bin/kafka-server-start.sh config/server-1.properties &

bin/kafka-server-start.sh config/server-2.properties &

二、集群验证

1、创建一个topic
现在创建一个复制因子为 3 的新主题：

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 1 --topic my-replicated-topic

查看该主题详情：

Leader 1指明集群的主节点是broker.id为1的节点，其负责给定分区的所有读取和写入。

Replicas: 1,2,0指明该主题对应的复制节点应该有哪些，此处表明集群中的三个节点都会同步保存该topic中的记录。

Isr:1,2,0指明该主题对应的复制节点有哪些在实际正常运行，此处表明集群中的三个节点都在正常同步保存该topic中的记录。

2、发送消息到topic

# 发送消息
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic my-replicated-topic

# 打开另一个会话窗口，接收消息
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic my-replicated-topic

干掉当前主节点：

如果主节点挂了，集群会从从节点中另外选择一个节点来作为主节点，保障集群的正常工作。因此仍然可以通过上面的发送消息命令来发送消息，只不过broker-list需要包含有仍在工作的节点。