Mac 本地搭建 Kafka+ZooKeeper

从 https://mirrors.tuna.tsinghua.edu.cn/apache/zookeeper/stable/ 下载 ZooKeeper 并解压缩。

cd zookeeper-3.4.10
cp -rf conf/zoo_sample.cfg conf/zoo.cfg
cd conf
nano zoo.cfg

修改配置文件，将

# the directory where the snapshot is stored.
# do not use /tmp for storage, /tmp here is just
# example sakes.
dataDir=/tmp/zookeeper

改成

dataDir=/Users/XXX/Documents/zkdata #这个目录是预先创建的

然后启动 ZooKeeper

$ ./zookeeper-3.4.10/bin/zkServer.sh start
ZooKeeper JMX enabled by default
Using config: /Users/XXX/Documents/XXX/zookeeper-3.4.10/bin/../conf/zoo.cfg
Starting zookeeper ... STARTED

查看

$ bin/zkServer.sh status
ZooKeeper JMX enabled by default
Using config: /Users/XXX/Documents/XXX/zookeeper-3.4.10/bin/../conf/zoo.cfg
Mode: standalone

$ ps ax | grep zookeeper
30735 s001  S      0:00.99 /Library/Java/JavaVirtualMachines/jdk1.8.0_144.jdk/Contents/Home/bin/java
-Dzookeeper.log.dir=.
-Dzookeeper.root.logger=INFO,CONSOLE
-cp /Users/XXX/Documents/XXX/zookeeper-3.4.10/bin/../build/classes:/Users/XXX/Documents/XXX/zookeeper-3.4.10/bin/../build/lib/*.jar:/Users/XXX/Documents/XXX/zookeeper-3.4.10/bin/../lib/slf4j-log4j12-1.6.1.jar:/Users/XXX/Documents/XXX/zookeeper-3.4.10/bin/../lib/slf4j-api-1.6.1.jar:/Users/XXX/Documents/XXX/zookeeper-3.4.10/bin/../lib/netty-3.10.5.Final.jar:/Users/XXX/Documents/XXX/zookeeper-3.4.10/bin/../lib/log4j-1.2.16.jar:/Users/XXX/Documents/XXX/zookeeper-3.4.10/bin/../lib/jline-0.9.94.jar:/Users/XXX/Documents/XXX/zookeeper-3.4.10/bin/../zookeeper-3.4.10.jar:/Users/XXX/Documents/XXX/zookeeper-3.4.10/bin/../src/java/lib/*.jar:/Users/XXX/Documents/XXX/zookeeper-3.4.10/bin/../conf: -Dcom.sun.management.jmxremote
-Dcom.sun.management.jmxremote.local.only=false org.apache.zookeeper.server.quorum.QuorumPeerMain /Users/XXX/Documents/XXX/zookeeper-3.4.10/bin/../conf/zoo.cfg

下载 Kafka，解压缩，启动

$ cd kafka_2.12-1.0.0
$ bin/kafka-server-start.sh config/server.properties &
...
...
[2017-11-04 10:14:28,410] INFO Kafka version : 1.0.0 (org.apache.kafka.common.utils.AppInfoParser)
[2017-11-04 10:14:28,410] INFO Kafka commitId : aaa7af6d4a11b29d (org.apache.kafka.common.utils.AppInfoParser)
[2017-11-04 10:14:28,412] INFO [KafkaServer id=0] started (kafka.server.KafkaServer)

打开新的终端，创建 Topic

$ bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
Created topic "test".
$ bin/kafka-topics.sh --list --zookeeper localhost:2181
test

启动消费者

$ bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test

新开终端，启动生产者

$ bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

在生产者的终端输入任何消息，几秒之后会在消费者的终端显示。

创建两个新的 broker

$ cd /Users/XXX/Documents/XXX/kafka_2.12-1.0.0
$ cp config/server.properties config/server-1.properties
$ cp config/server.properties config/server-2.properties

修改配置文件 config/server-1.properties

broker.id=1
listeners=PLAINTEXT://localhost:9093
log.dir=/tmp/kafka-logs-1

修改配置文件 config/server-2.properties

broker.id=2
listeners=PLAINTEXT://localhost:9094
log.dir=/tmp/kafka-logs-2

启动

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

创建新 Topic

$ bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 1 --topic test
Created topic "test".

$ bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic test
Topic:test	PartitionCount:1	ReplicationFactor:3	Configs:
Topic: test	Partition: 0	Leader: 2	Replicas: 2,0,1	Isr: 2,0,1

向新建的 Topic 发送消息

$ bin/kafka-console-producer.sh --broker-list localhost:9094 --topic test1
>message1
>message2

启动消费者接收消息

$ bin/kafka-console-consumer.sh --bootstrap-server localhost:9094 --from-beginning --topic test1
message1
message2

现在来测试下容错能力，因为 broker 2是leader，试着把它干掉

$ ps aux | grep server-2.properties
40255 0:18.41 /Library/Java/JavaVirtualMachines/jdk1.8.0_144.jdk/Contents/Home/bin/java ...

$ kill -9 40255

现在 leader 已经换成了 broker 0

$ bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic test1
Topic:test1	PartitionCount:1	ReplicationFactor:3	Configs:
Topic: test1	Partition: 0	Leader: 0	Replicas: 2,0,1	Isr: 0,1

同时消费者终端出现如下提示：

[2017-11-04 11:55:38,420] WARN [Consumer clientId=consumer-1, groupId=console-consumer-92757] Connection to node 2 could not be established. Broker may not be available. (org.apache.kafka.clients.NetworkClient)

继续从生产者终端发送消息，发现消费者终端仍然能接收到消息。

参考：http://kafka.apache.org/quickstart