「不要屈服，不要淡化，不要使它看来合逻辑，不要依据潮流而修改你的灵魂。相反的，狠狠的追随你最强烈的喜好之事。」
“Don't bend; don't water it down; don't try to make it logical; don't edit your own soul according to the fashion. Rather, follow your most intense obsessions mercilessly.”
― Franz Kafka

昨天新增在单一个 broker 上新增了一个 topic，今天将要带大家新增多个 broker，并且真的的实现 kafka 持久化。

Step 1: 设定并启动三个 broker

1.1 调整 broker 设定档

• 设定多个broker
  • 进入到 Kafka 资料夹中，将 server.properties 设定档复制三份

$ cd /usr/local/etc/kafka
$ cp server.properties server-1.properties
$ cp server.properties server-2.properties
$ cp server.properties server-3.properties

• 依序修改设定档
  • 我们要在同一台机器运作三个 broker，所以必须修改 port
  • 每个 broker 的 id 必须不同，这边也须修改
  • log.dirs 就是让讯息所谓持久化存放的位置，这边也依照 broker 区分
  • 修改 server-1.properties

broker.id=0
listeners=PLAINTEXT://localhost:9092
log.dirs=/usr/local/var/lib/kafka-logs/broker1

• 修改 server-2.properties

broker.id=1
listeners=PLAINTEXT://localhost:9093
log.dirs=/usr/local/var/lib/kafka-logs/broker2

• 修改 server-3.properties

broker.id=2
listeners=PLAINTEXT://localhost:9094
log.dirs=/usr/local/var/lib/kafka-logs/broker3

1.2 启动三个 Broker

接下来，开启三个终端机，分别启动 broker1, broker2, broker3，这边不用背景启动，方便看 log 熟悉 broker server 实际上在运行什麽

$ cd  /usr/local/Cellar/kafka/2.8.0/libexec
$ kafka-server-start /usr/local/etc/kafka/server-1.properties
$ kafka-server-start /usr/local/etc/kafka/server-2.properties
$ kafka-server-start /usr/local/etc/kafka/server-3.properties

Step 2: 新增 Topic

2.1 这边要用kafka-topics.sh去新增 topic

$ kafka-topics --create --zookeeper 127.0.0.1:2181 --replication-factor 2 --partitions 3 --topic topicWithThreeBroker

这边跟昨天创建单一 broker 的 topic 有差别的参数是 replication-factor 跟 partitions

• --partitions 3 => 在 topic 下建立三个分区
• --replication-factor 2 => 意思是将每一个分区复制到其中两个 broker 上

这样创建的好处是什麽？
就算其中一个broker挂掉了，仍然可以从其他两个broker获取资料，但如果坏掉2个broker还是会掉资料，这样的做法，只可以容忍1个broker坏掉

实际上的log目录结构是这样，可以清楚地看出这样的配置是可以容忍N-1个 broker 坏掉：

* broker0
    * topic1-0
    * topic1-1
* broker1
    * topic1-1
    * topic1-2
* broker2
    * topic1-2
    * topic1-0

2.2 查看一下topic实际上是如何分配的

$ kafka-topics --describe --zookeeper 127.0.0.1:2181 --topic topicWithThreeBroker

Topic: topicWithThreeBroker	TopicId: BAocHAwHR_STmwAUlI3YMw	PartitionCount: 3	ReplicationFactor: 2	Configs:
	Topic: topicWithThreeBroker	Partition: 0	Leader: 1	Replicas: 1,0	Isr: 1,0
	Topic: topicWithThreeBroker	Partition: 1	Leader: 2	Replicas: 2,1	Isr: 2,1
	Topic: topicWithThreeBroker	Partition: 2	Leader: 0	Replicas: 0,2	Isr: 0,2

参数介绍：
Partition: 0 => 代表 Topic topicWithThreeBroker 下编号为0的 Parition
Leader: 1 => 代表 partition 0 的 partition leader 在 broker1 上
=> 每个分区都有一个 leader，通常leader不会集中在同一个节点上
=> 这个 leader 会负责该 partition 内所有的读写，而其他的 broker 上的 replica(follower)，会去消费、同步该 leader 的资料
=> 如果这个 leader 挂掉了，那麽会从 Isr 中，选取第一个(先到先赢机制)成为该 partition 新的 leader
Replicas 1,0 => 代表该分区有副本在 broker1 和 broker0 上
Isr: 1, 0 => 代表我们当前可以在 broker1 和 broker0 上访问该 partition0 的资料
=> 如果broker 0挂掉，Isr 会从 Isr: 1,0 变成 Isr: 1
=> Isr 的全名为in-sync replica，也就是已同步的副本

2.3 模拟其中一个 broker 坏掉後再恢复的情况

这边将broker0 ctrl+c关掉，模拟少一个broker的情况，同样去下指令

$ kafka-topics --describe --zookeeper 127.0.0.1:2181 --topic topicWithThreeBroker

Topic: topicWithThreeBroker	TopicId: BAocHAwHR_STmwAUlI3YMw	PartitionCount: 3	ReplicationFactor: 2	Configs:
	Topic: topicWithThreeBroker	Partition: 0	Leader: 1	Replicas: 1,0	Isr: 1
	Topic: topicWithThreeBroker	Partition: 1	Leader: 2	Replicas: 2,1	Isr: 2,1
	Topic: topicWithThreeBroker	Partition: 2	Leader: 2	Replicas: 0,2	Isr: 2

这边可以看到 Parition0 的 Isr 剩下 broker1、Partition2 的 Isr 剩下 broker2
而原本 Partition2 的 Leader 因为 Broker0 坏掉，会从 Isr 中选取，因此 Leader 变成 Broker2

把 broker0 重新启动

$ kafka-server-start /usr/local/etc/kafka/server-1.properties

再下一次指令查看 Topic topicWithThreeBroker 的分配状况

$ kafka-topics --describe --zookeeper 127.0.0.1:2181 --topic topicWithThreeBroker

Topic: topicWithThreeBroker	TopicId: BAocHAwHR_STmwAUlI3YMw	PartitionCount: 3	ReplicationFactor: 2	Configs:
	Topic: topicWithThreeBroker	Partition: 0	Leader: 1	Replicas: 1,0	Isr: 1,0
	Topic: topicWithThreeBroker	Partition: 1	Leader: 2	Replicas: 2,1	Isr: 2,1
	Topic: topicWithThreeBroker	Partition: 2	Leader: 2	Replicas: 0,2	Isr: 2,0

现在可以看到 Partition0 和 Parition2 的 Isr 都有恢复了
Partition: 0 Isr: 2,0
Partition: 2 Isr: 1,0
也可以观察到 Partition2 的 leader 没有自动回归到 broker0 上，还是在broker2上，这部分会在之後介绍如何重新分配 Partition Leader

Step 3: Broker 常用设定

Kafka cluster 会自动侦测是否有 broker 关闭或是失败，然後帮那个 broker 上的 partition 重新选举 leader，这个机制会在 broker 服务器当掉、为了要维护而关掉、config档案更改时生效。
Kafka 支援一个优雅的关闭机制去让 server 停止运作，而非直接kill掉，这个机制会做两件事：

1. 会将所有的log同步到磁碟中，避免重启时还要去做log复原的动作，log复原需要时间，所以这个机制会加快 broker 重启的速度。
2. 在将 broker 关闭之前，这个机制会移动在这个 broker 上是 leader 的 partition 到其他的 replica，这会加速 leader 转移的时间、并将 partition 不可用的时间下降到几毫秒(milliseconds); 除了hard kill之外，当 server 关闭或是失败时触发这个机制。

使用方式：
在档案位置/usr/local/etc/kafka/server-1.properties加上设定controlled.shutdown.enable=true

需要注意的是 controlled shutdown 只会在broker上所有 partition 都有一个以上的 replicas 时才会成功。