Kafka 副本和集群
在生产环境中,Kafka 总是以“集群+分区”方式运行的,以保证可靠性和性能。下面是一个3副本的 Kafka 集群实例。
首先,需要启动3个 Kafka Broker,Broker 的配置文件分别如下,
broker.id=0
listeners=PLAINTEXT://192.168.232.23:9092
log.dirs=/tmp/kafka-logs
broker.id=1
listeners=PLAINTEXT://192.168.232.23:9093
log.dirs=/tmp/kafka-logs-1
broker.id=1
listeners=PLAINTEXT://192.168.232.23:9094
log.dirs=/tmp/kafka-logs-2
虽然每个 Broker 只配置了一个端口,实际上,Kafka 会多占用一个,可能是用来 Broker 之间的复制的。另外,3个 Broker 都配置了,
zookeeper.connect=localhost:2181
delete.topic.enable=true
在同一个 Zookeeper 上的 Broker 会被归类到一个集群中。注意,这些配置中并没有指定哪一个 Broker 是主节点,哪些 Broker 是从节点,Kafka 采用的办法是从可选的 Broker 中,选出每个分区的 Leader。也就是说,对某个 Topic 来说,可能0节点是 Leader,另外一些 Topic,可能1节点是 Leader;甚至,如果 topic1 有2个分区的话,分区1的 Leader 是0节点,分区2的 Leader 是1节点。
这种对等的设计,对于故障恢复是十分有用的,在节点崩溃的时候,Kafka 会自动选举出可用的从节点,将其升级为主节点。在崩溃的节点恢复,加入集群之后,Kafka 又会将这个节点加入到可用节点,并自动选举出新的主节点。
实验如下,先新建一个3副本,2分区的 Topic,
1bin/kafka-topics.sh --create --zookeeper 192.168.232.23:2181 --replication-factor 3 --partitions 2 --topic topic1
初始状况下,topic1 的状态如下,
$ bin/kafka-topics.sh --describe --zookeeper 192.168.232.23:2181 --topic topic1
Topic:topic1 PartitionCount:2 ReplicationFactor:3 Configs:
Topic: topic1 Partition: 0 Leader: 0 Replicas: 0,1,2 Isr: 0,1,2
Topic: topic1 Partition: 1 Leader: 1 Replicas: 1,2,0 Isr: 1,2,0
对于上面的输出,即使没有文档,也可以看懂大概:topic1 有2个分区,Partition 0 和 Partition 1,Leader 分别在 Broker 0 和 1。Replicas 表示副本在哪些 Broker 上,Isr(In-Sync Replicas)表示处于同步状态中的 Broker,如果有 Broker 宕机了,那么 Replicas 不会变,但是 Isr 会仅显示没有宕机的 Broker,详见下面的实验。
然后分2个线程,运行之前写的 Producer 和 Consumer 的示例代码,Producer 采用异步发送,消息采用同步复制。在有消息传送的情况下,kill -9
停掉其中2个 Broker(Broker 0 和 Broker 1),模拟突然宕机。此时,topic1 状态如下,
$ bin/kafka-topics.sh --describe --zookeeper 192.168.232.23:2181 --topic topic1
Topic:topic1 PartitionCount:2 ReplicationFactor:3 Configs:
Topic: topic1 Partition: 0 Leader: 2 Replicas: 0,1,2 Isr: 2
Topic: topic1 Partition: 1 Leader: 2 Replicas: 1,2,0 Isr: 2
可见,Kafka 已经选出了新的 Leader,消息传送没有中断。接着再启动被停掉的那两个 Broker,并查看 topic1 的状态,如下,
$ bin/kafka-topics.sh --describe --zookeeper 192.168.232.23:2181 --topic topic1
Topic:topic1 PartitionCount:2 ReplicationFactor:3 Configs:
Topic: topic1 Partition: 0 Leader: 2 Replicas: 0,1,2 Isr: 2,1,0
Topic: topic1 Partition: 1 Leader: 2 Replicas: 1,2,0 Isr: 2,1,0
$ bin/kafka-topics.sh --describe --zookeeper 192.168.232.23:2181 --topic topic1
Topic:topic1 PartitionCount:2 ReplicationFactor:3 Configs:
Topic: topic1 Partition: 0 Leader: 2 Replicas: 0,1,2 Isr: 2,1,0
Topic: topic1 Partition: 1 Leader: 1 Replicas: 1,2,0 Isr: 2,1,0
可以发现, 有一个短暂的时间,topic1 的两个分区的 Leader 都是 Broker 2,但是在 Kafka 重新选举之后,分区1的 Leader 变为 Broker 1。说明 Kafka 倾向于用不同的 Broker 做分区的 Leader,这样更能达到负载均衡的效果。
再来看看 Producer 和 Consumer 的日志,下面这个片段是2个 Broker 宕机前后的日志,
......
Send message: (00439, Message_00439) at offset 217 to partition(0) in
3 ms
Received message: (00438, Message_00438) at offset 216
Send message: (00440, Message_00440) at offset 218 to partition(0) in
5 ms
Send message: (00441, Message_00441) at offset 221 to partition(1) in
5 ms
Received message: (00441, Message_00441) at offset 221
Received message: (00439, Message_00439) at offset 217
Send message: (00442, Message_00442) at offset 222 to partition(1) in
5 ms
Send message: (00443, Message_00443) at offset 219 to partition(0) in
3 ms
Received message: (00440, Message_00440) at offset 218
Received message: (00443, Message_00443) at offset 219
org.apache.kafka.common.errors.NetworkException: The server disconnected before a response was received.
org.apache.kafka.common.errors.NetworkException: The server disconnected before a response was received.
org.apache.kafka.common.errors.NetworkException: The server disconnected before a response was received.
org.apache.kafka.common.errors.NetworkException: The server disconnected before a response was received.
org.apache.kafka.common.errors.NetworkException: The server disconnected before a response was received.
org.apache.kafka.common.errors.NetworkException: The server disconnected before a response was received.
org.apache.kafka.common.errors.NetworkException: The server disconnected before a response was received.
Received message: (00442, Message_00442) at offset 222
Send message: (00452, Message_00452) at offset 223 to partition(1) in
7492 ms
Send message: (00454, Message_00454) at offset 224 to partition(1) in
7485 ms
Send message: (00455, Message_00455) at offset 225 to partition(1) in
7482 ms
Send message: (00458, Message_00458) at offset 226 to partition(1) in
7473 ms
Send message: (00460, Message_00460) at offset 227 to partition(1) in
7467 ms
Send message: (00461, Message_00461) at offset 228 to partition(1) in
7465 ms
Send message: (00462, Message_00462) at offset 229 to partition(1) in
7462 ms
Send message: (00463, Message_00463) at offset 230 to partition(1) in
7459 ms
Send message: (00464, Message_00464) at offset 231 to partition(1) in
7456 ms
Send message: (00465, Message_00465) at offset 232 to partition(1) in
7453 ms
......
Send message: (01103, Message_01103) at offset 543 to partition(1) in
5478 ms
Received message: (00631, Message_00631) at offset 310
Received message: (00633, Message_00633) at offset 311
Send message: (00451, Message_00451) at offset 220 to partition(0) in
7525 ms
Received message: (00634, Message_00634) at offset 312
Send message: (00453, Message_00453) at offset 221 to partition(0) in
7518 ms
Received message: (00639, Message_00639) at offset 313
Send message: (00456, Message_00456) at offset 222 to partition(0) in
7509 ms
Received message: (00641, Message_00641) at offset 314
Send message: (00457, Message_00457) at offset 223 to partition(0) in
7506 ms
Received message: (00643, Message_00643) at offset 315
......
出现错误的时候,Producer 抛出了 NetworkException
异常。其中有3589条 Received 日志,3583条 Send 日志,7条 NetworkException
异常日志,发送消息的最大序号是3590,接收消息的最大序号是3589,有以下几个值得注意的地方,
- 宕机之前,消息的接收并不是顺序的,这是因为 topic1 有2个分区,Kafka 只保证分区上的有序;
- 宕机之后,出现了长段的发送日志而没有接收日志,说明 Kafka 此时正在选举,选举的过程会阻塞消费者;
- 从接收消息的条数和序号来看,所有的消息都收到了,没有丢(没有收到3590的消息可能是因为强制退出 client 进程的原因),发送的过程的7个异常应该只是虚警,7条异常对应序号444~450,3583条 Send 消息再加上这7条,与总消息3590条一致;
从这个实验中,可以看到,虽然 Kafka 不保证消息重复发送,但是却在尽量保证没有消息被重复发送,可能我的实验场景还不够极端,没有做出消息重复的情况。
如之前所说,如果要保持完全顺序性,需要使用单分区;如果要避免抛出 NetworkException
异常,就使用 Producer 同步发送。下面,我们重做上面的例子,不同之处是使用单分区和 Producer 同步发送,截取一段 Broker 宕机时的日志如下,
......
Sent message: (118, Message_00118)
Received message: (00118, Message_00118) at offset 117
Received message: (00119, Message_00119) at offset 118
Sent message: (119, Message_00119)
Sent message: (120, Message_00120)
Received message: (00120, Message_00120) at offset 119
Sent message: (121, Message_00121)
Received message: (00121, Message_00121) at offset 120
Sent message: (122, Message_00122)
Sent message: (123, Message_00123)
Sent message: (124, Message_00124)
Sent message: (125, Message_00125)
Sent message: (126, Message_00126)
Sent message: (127, Message_00127)
......
可见,由于采用同步发送,Broker 宕机并没有造成抛出异常,另外,由于使用单分区,顺序性也得到了保证,全局没有出现乱序的情况。
综上,是否使用多分区更多的是对顺序性的要求,而使用 Producer 同步发送还是异步发送,更多是出于重复消息的考虑,如果异步发送抛出异常,在保证不丢消息的前提下,势必要重发消息,这就会导致收到重复消息。多分区和 Producer 异步发送,会带来性能的提升,但是也会引入非顺序性,重复消息等问题,如何取舍要看应用的需求。
Kafka 最佳实践
Kafka 在一些应用场景中,有一些前人总结的最佳实践 8 9。对最佳实践,我的看法是,对于自己比较熟悉,有把握的部分,可以按自己的步骤进行;对一些自己不清楚的领域,可以借鉴其中的一些内容,至少不会错的特别厉害。有文章10说,Kafka 在分区比较多的时候,相应时间会变长,这个现象值得在实践中注意。
后记
在 Kafka 与 RocketMQ 的对比中,RocketMQ 的一个核心功能就是可以支持同步刷盘,此时,即使突然断电,也可以保证消息不丢;而 Kafka 采用的是异步刷盘,即使返回写入成功,也只是写入缓冲区成功,并非已经持久化。因此,如果出现断电或 kill -9
的情况,Kafka 内存中的消息可能丢失。另外,同步刷盘的效率是比较低下的,一般生产中估计也不会使用,可以用优雅关闭的方式来关闭进程。如果不考虑这些极端情况的话,Kafka 基本是一个很可靠的消息中间件。
欢迎工作一到五年的Java工程师朋友们加入Java架构开发: 854393687
群内提供免费的Java架构学习资料(里面有高可用、高并发、高性能及分布式、Jvm性能调优、Spring源码,MyBatis,Netty,Redis,Kafka,Mysql,Zookeeper,Tomcat,Docker,Dubbo,Nginx等多个知识点的架构资料)合理利用自己每一分每一秒的时间来学习提升自己,不要再用”没有时间“来掩饰自己思想上的懒惰!趁年轻,使劲拼,给未来的自己一个交代!