架构师之路-如何建立高可用消息中间件kafka？ _经验知识

1 概述
Kafka与传统消息系统相比。有以下不同：
它被设计为一个分布式系统。易于向外扩展；
它同时为发布和订阅提供高吞吐量；
它支持多订阅者。当失败时能自动平衡消费者；
它将消息持久化到磁盘。因此可用于批量消费。例如ETL以及实时应用程序。
Kafka凭借着自身的优势。越来越受到互联网企业的青睐。唯品会也采用Kafka作为其内部核心消息引擎之一。Kafka作为一个商业级消息中间件。消息可靠性的重要性可想而知。如何确保消息的精确传输？如何确保消息的准确存储？如何确保消息的正确消费？这些都是需要考虑的问题。本文首先从Kafka的架构着手。先了解下Kafka的基本原理。然后通过对kakfa的存储机制、复制原理、同步原理、可靠性和持久性保证等等一步步对其可靠性进行分析。最后通过benchmark来增强对Kafka高可靠性的认知。
2 Kafka体系架构

文章插图
如上图所示。一个典型的Kafka体系架构包括若干Producer（可以是服务器日志。业务数据。页面前端产生的page view等等）。若干broker（Kafka支持水平扩展。一般broker数量越多。集群吞吐率越高）。若干Consumer (Group) 。以及一个Zookeeper集群。Kafka通过Zookeeper管理集群配置。选举leader 。以及在consumer group发生变化时进行rebalance 。Producer使用push(推)模式将消息发布到broker 。Consumer使用pull(拉)模式从broker订阅并消费消息。
名词解释：

文章插图
2.1 Topic & Partition
一个topic可以认为一个一类消息。每个topic将被分成多个partition 。每个partition在存储层面是append log文件。任何发布到此partition的消息都会被追加到log文件的尾部。每条消息在文件中的位置称为offset(偏移量) 。offset为一个long型的数字。它唯一标记一条消息。每条消息都被append到partition中。是顺序写磁盘。因此效率非常高（经验证。顺序写磁盘效率比随机写内存还要高。这是Kafka高吞吐率的一个很重要的保证）。

文章插图
每一条消息被发送到broker中。会根据partition规则选择被存储到哪一个partition 。如果partition规则设置的合理。所有消息可以均匀分布到不同的partition里。这样就实现了水平扩展。（如果一个topic对应一个文件。那这个文件所在的机器I/O将会成为这个topic的性能瓶颈。而partition解决了这个问题）。在创建topic时可以在$KAFKA_HOME/config/server.properties中指定这个partition的数量（如下所示）。当然可以在topic创建之后去修改partition的数量。

文章插图
在发送一条消息时。可以指定这个消息的key 。producer根据这个key和partition机制来判断这个消息发送到哪个partition 。partition机制可以通过指定producer的partition.class这一参数来指定。该class必须实现kafka.producer.Partitioner接口。
有关Topic与Partition的更多细节。可以参考下面的“Kafka文件存储机制”这一节。
3 高可靠性存储分析
Kafka的高可靠性的保障来源于其健壮的副本（replication）策略。通过调节其副本相关参数。可以使得Kafka在性能和可靠性之间运转的游刃有余。Kafka从0.8.x版本开始提供partition级别的复制,replication的数量可以在$KAFKA_HOME/config/server.properties中配置（default.replication.refactor）。
这里先从Kafka文件存储机制入手。从最底层了解Kafka的存储细节。进而对其的存储有个微观的认知。之后通过Kafka复制原理和同步方式来阐述宏观层面的概念。最后从ISR 。HW 。leader选举以及数据可靠性和持久性保证等等各个维度来丰富对Kafka相关知识点的认知。
3.1 Kafka文件存储机制
Kafka中消息是以topic进行分类的。生产者通过topic向Kafka broker发送消息。消费者通过topic读取数据。然而topic在物理层面又能以partition为分组。一个topic可以分成若干个partition 。那么topic以及partition又是怎么存储的呢？partition还可以细分为segment 。一个partition物理上由多个segment组成。那么这些segment又是什么呢？下面我们来一一揭晓。
为了便于说明问题。假设这里只有一个Kafka集群。且这个集群只有一个Kafka broker 。即只有一台物理机。在这个Kafka broker中配置（$KAFKA_HOME/config/server.properties中）log.dirs=/tmp/kafka-logs 。以此来设置Kafka消息文件存储目录。与此同时创建一个topic：topic_vms_test 。partition的数量为4（$KAFKA_HOME/bin/kafka-topics.sh --create --zookeeper localhost:2181 --partitions 4 --topic topic_vms_test --replication-factor 4）。那么我们此时可以在/tmp/kafka-logs目录中可以看到生成了4个目录：