配置jdk
[[email protected] mnt]# java -version
java version "1.8.0_102"
Java(TM) SE Runtime Environment (build 1.8.0_102-b14)
Java HotSpot(TM) 64-Bit Server VM (build 25.102-b14, mixed mode)

解壓zookeeper

解壓kafka

[[email protected] kafka]# vim /etc/hosts  解析本地主機名

[[email protected] zookeeper]# cp conf/zoo_sample.cfg conf/zoo_cfg   生成zookeeper配置檔案

[

[email protected] zookeeper]# ./bin/zkServer.sh start    啟動zookeeper
JMX enabled by default
Using config: /mnt/zookeeper/bin/../conf/zoo.cfg
Starting zookeeper ... STARTED

[[email protected] kafka]# nohup ./bin/kafka-server-start.sh config/server.properties >> /mnt/kafka.log &   啟動kafka     

[[email protected] kafka]# ./bin/kafka-topics.sh  --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic kafkatest
[[email protected] kafka]# ./bin/kafka-topics.sh  --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
建立2個主題


[[email protected] kafka]# bin/kafka-topics.sh --list --zookeeper localhost:2181     檢視執行的主題
kafkatest
test
[

[email protected] kafka]#

在終端執行生產者
[[email protected] kafka]# ./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
wocao
wocao!
我曹


另外開啟一個終端執行消費者（卡夫卡有一個命令，是消費者將轉儲訊息到標準輸出。）
[[email protected] kafka]# bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning
wocao
wocao!
我曹

擴大我們的叢集到三個節點（仍然在我們的本地機器）。
複製出另外兩個配置檔案
[[email protected] kafka]# cp config/server.properties config/server_1.properties
[[email protected] kafka]# cp config/server.properties config/server_2.properties

修改server_1和server_2配置檔案（列出三個配置檔案的內容）
server.properties為預設檔案
[[email protected] kafka]# grep -v "^$" config/server.properties | grep -v "^#"
broker.id=0
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/tmp/kafka-logs
num.partitions=1
num.recovery.threads.per.data.dir=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=localhost:2181
zookeeper.connection.timeout.ms=6000

server_1.properties內容
[[email protected] kafka]# grep -v "^$" config/server_1.properties | grep -v "^#"
broker.id=1
listeners=PLAINTEXT://:9093
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/tmp/kafka-logs_1
num.partitions=1
num.recovery.threads.per.data.dir=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=localhost:2181
zookeeper.connection.timeout.ms=6000

server_2.properties內容
[[email protected] kafka]# grep -v "^$" config/server_2.properties | grep -v "^#"
broker.id=2
listeners=PLAINTEXT://:9094
num.network.threads=3
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
socket.request.max.bytes=104857600
log.dirs=/tmp/kafka-logs_2
num.partitions=1
num.recovery.threads.per.data.dir=1
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=localhost:2181
zookeeper.connection.timeout.ms=6000

啟動另外兩個節點
[[email protected] kafka]# nohup ./bin/kafka-server-start.sh config/server_1.properties >> /mnt/kafka_1.log &
[1] 4451
[[email protected] kafka]# nohup: 忽略輸入重定向錯誤到標準輸出端
[[email protected] kafka]# nohup ./bin/kafka-server-start.sh config/server_2.properties >> /mnt/kafka_2.log &
[2] 4744
[[email protected] kafka]# nohup: 忽略輸入重定向錯誤到標準輸出端


現在建立一個新的主題，複製係數為3
[[email protected] kafka]# ./bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 1 --topic kafka_3_node

檢視主題kafka_3_node
[[email protected] kafka]# bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic kafka_3_node
Topic:kafka_3_node    PartitionCount:1    ReplicationFactor:3    Configs:
    Topic: kafka_3_node    Partition: 0    Leader: 1    Replicas: 1,0,2    Isr: 1,0,2
說明：
Topic（主題）：kafka_3_node
PartitionCount（分割槽數）：1
Leader（領導）在1節點上
replicas（副本）在1,0,2節點，不管node死活，只是列出資訊而已.
isr” 工作中的複製節點的集合. 也就是活的節點的集合.

再看看主題test
[[email protected] kafka]# bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic test
Topic:test    PartitionCount:1    ReplicationFactor:1    Configs:
    Topic: test    Partition: 0    Leader: 0    Replicas: 0    Isr: 0

傳送一些新的訊息kafka_3_node
[[email protected] kafka]# bin/kafka-console-producer.sh --broker-list localhost:9092 --topic kafka_3_node
woqu
我去
我去去去去去去去


用另外一個終端接收
[[email protected] kafka]# bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic kafka_3_node
woqu
我去
我去去去去去去去


幹掉leader,也就是Broker1
[[email protected] kafka]# kill 4451

再看一下kafka_3_node
[[email protected] kafka]# bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic kafka_3_node
Topic:kafka_3_node    PartitionCount:1    ReplicationFactor:3    Configs:
    Topic: kafka_3_node    Partition: 0    Leader: 0    Replicas: 1,0,2    Isr: 0,2

對比一下之前的
[[email protected] kafka]# bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic kafka_3_node
Topic:kafka_3_node    PartitionCount:1    ReplicationFactor:3    Configs:
    Topic: kafka_3_node    Partition: 0    Leader: 1    Replicas: 1,0,2    Isr: 1,0,2

再看一下消費者，訊息是沒有丟失的
[[email protected] kafka]# bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic kafka_3_node
woqu
我去
我去去去去去去去