Kafka 集群

创建集群

$ docker run -d --name kafka1 -p 9093:9092 \
--net zookeeper \
--env KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181 \
--env KAFKA_LISTENERS=PLAINTEXT://kafka1:9092 \
--env KAFKA_BROKER_ID=1 \
--env KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://kafka1:9092 \
--hostname kafka1 \
wurstmeister/kafka:2.12-2.3.1

$ docker run -d --name kafka2 -p 9094:9092 \
--net zookeeper \
--env KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181 \
--env KAFKA_LISTENERS=PLAINTEXT://kafka2:9092 \
--env KAFKA_BROKER_ID=2 \
--env KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://kafka2:9092 \
--hostname kafka2 \
wurstmeister/kafka:2.12-2.3.1

创建Topic

• 创建一个replication-factor为3的topic

$ kafka-topics.sh --create --zookeeper zookeeper:2181 --replication-factor 3 --partitions 1 --topic test-replicated-topic

查看Topic

• 运行describe topics命令。

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

• leader：负责给定Partition的所有读写的节点，每个节点都可能成为Partition随机选择的leader
• replicas：复制此Partition日志的节点列表，无论它们是leader还是当前处于存活状态。
• isr：是一组 “in-sync” replicas，这是replicas列表的一个子集，它当前处于存活状态，并补充leader

Topic发布信息

$ kafka-console-producer.sh --broker-list kafka:9092 --topic test-replicated-topic
> test message 1
> test message 2

接收信息

$ kafka-console-consumer.sh --bootstrap-server kafka:9092 --from-beginning --topic test-replicated-topic
test message 1
test message 2

容错性

• 关闭node0节点，原leader已被替换成它的flowers中的其中一个，并且 node 0 不在 in-sync replica 集合当中。

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

• 尽管原leader已逝，但原来消息依然可以通过其他node接收。

$ kafka-console-consumer.sh --bootstrap-server kafka2:9092 --from-beginning --topic test-replicated-topic
test message 1
test message 2
test message 3