一 Kafka 都有哪些特点?
- 高吞吐量、低延迟:kafka每秒可以处理几十万条消息,它的延迟最低只有几毫秒,每个topic可以分多个partition, consumer group 对partition进行consume操作。
- 可扩展性:kafka集群支持热扩展
- 持久性、可靠性:消息被持久化到本地磁盘,并且支持数据备份防止数据丢失
- 容错性:允许集群中节点失败(若副本数量为n,则允许n-1个节点失败)
- 高并发:支持数千个客户端同时读写
二 Kafka的应用场景
- 日志收集:一个公司可以用Kafka可以收集各种服务的log,通过kafka以统一接口服务的方式开放给各种consumer,例如hadoop、HBase、Solr等。
- 消息系统:解耦和生产者和消费者、缓存消息等。
- 用户活动跟踪:Kafka经常被用来记录web用户或者app用户的各种活动,如浏览网页、搜索、点击等活动,这些活动信息被各个服务器发布到kafka的topic中,然后订阅者通过订阅这些topic来做实时的监控分析,或者装载到hadoop、数据仓库中做离线分析和挖掘。
- 运营指标:Kafka也经常用来记录运营监控数据。包括收集各种分布式应用的数据,生产各种操作的集中反馈,比如报警和报告。
- 流式处理:比如spark streaming和 Flink
三 Kafka 的设计架构
Kafka 架构分为以下几个部分
- Producer :消息生产者,就是向 kafka broker 发消息的客户端。
- Consumer :消息消费者,向 kafka broker 取消息的客户端。
- Topic :可以理解为一个队列,一个 Topic 又分为一个或多个分区。
- Consumer Group:这是 kafka 用来实现一个 topic 消息的广播(发给所有的 consumer)和单播(发给任意一个 consumer)的手段。一个 topic 可以有多个 Consumer Group。
- Broker :一台 kafka 服务器就是一个 broker。一个集群由多个 broker 组成。一个 broker 可以容纳多个 topic。
- Partition:为了实现扩展性,一个非常大的 topic 可以分布到多个 broker上,每个 partition 是一个有序的队列。partition 中的每条消息都会被分配一个有序的id(offset)。将消息发给 consumer,kafka 只保证按一个 partition 中的消息的顺序,不保证一个 topic 的整体(多个 partition 间)的顺序。
- Offset:kafka 的存储文件都是按照 offset.kafka 来命名,用 offset 做名字的好处是方便查找。例如你想找位于 2049 的位置,只要找到 2048.kafka 的文件即可。当然 the first offset 就是 00000000000.kafka。
- leader 是 partition的 leader ,leader 有 replication 数,Isr 机制,leader在broker 上,加入broker 挂掉,会在isr 里 找一个作为leader
ACK有三种
0
意味着producer不等待broker同步完成的确认,继续发送下一条(批)信息
提供了最低的延迟。但是最弱的持久性,当服务器发生故障时,就很可能发生数据丢失。例如leader已经死亡,producer不知情,还会继续发送消息broker接收不到数据就会数据丢失
1
意味着producer要等待leader成功收到数据并得到确认,才发送下一条message。此选项提供了较好的持久性较低的延迟性。
Partition的Leader死亡,follwer尚未复制,数据就会丢失
-1
意味着producer得到follwer确认,才发送下一条数据
每个partition不再只有一个,而是有一个leader(红色)和多个replica(蓝色),生产者根据消息的topic和key值,确定了消息要发往哪个partition之后(假设是p1),会找到partition对应的leader(也就是broker2里的p1),然后将消息发给leader,leader负责消息的写入,并与其余的replica进行同步。
ACK = 0 时 发送一次 不论leader是否接收
ACK = 1 时,等待leader接收成功即可
ACK = -1 时 ,需等待leader将消息同步给follower 简单来说就是,producer只有收到分区内所有副本的成功写入的通知才认为推送消息成功了。
综上所述,为了保证数据的可靠性,我们最少需要配置一下几个参数:
- producer 级别:acks=all(或者 request.required.acks=-1),同时发生模式为同步 producer.type=sync
- topic 级别:设置 replication.factor>=3,并且 min.insync.replicas>=2;
- broker 级别:关闭不完全的 Leader 选举,即 unclean.leader.election.enable=false;
四 Kafka 分区的目的?
分区对于 Kafka 集群的好处是:实现负载均衡。分区对于消费者来说,可以提高并发度,提高效率。
五 Kafka 是如何做到消息的有序性?
kafka 中的每个 partition 中的消息在写入时都是有序的,而且单独一个 partition 只能由一个消费者去消费,可以在里面保证消息的顺序性。但是分区之间的消息是不保证有序的。
六 Kafka 的再均衡
在Kafka中,当有新消费者加入或者订阅的topic数发生变化时,会触发Rebalance(再均衡:在同一个消费者组当中,分区的所有权从一个消费者转移到另外一个消费者)机制,Rebalance顾名思义就是重新均衡消费者消费。Rebalance的过程如下:
第一步:所有成员都向coordinator发送请求,请求入组。一旦所有成员都发送了请求,coordinator会从中选择一个consumer担任leader的角色,并把组成员信息以及订阅信息发给leader。
第二步:leader开始分配消费方案,指明具体哪个consumer负责消费哪些topic的哪些partition。一旦完成分配,leader会将这个方案发给coordinator。coordinator接收到分配方案之后会把方案发给各个consumer,这样组内的所有成员就都知道自己应该消费哪些分区了。
所以对于Rebalance来说,Coordinator起着至关重要的作用
七 Kafka 在什么情况下会出现消息丢失
- 数据的一致性和可靠性
- 可靠性 是 通过分区 和 副本 ISR ,和ack机制 来保证数据的可靠性
- 一致性 是 High Water Mark 取决于 ISR 列表里面偏移量最小的分区,对应于上图的副本2,这个很类似于木桶原理。
消费者 只能看到 Message2 这条消息,保证了数据的一致性。
Kafka 复制机制
Kafka 主题中的每个分区都有一个预写日志(write-ahead log),我们写入 Kafka 的消息就存储在这里面。这里面的每条消息都有一个唯一的偏移量,用于标识它在当前分区日志中的位置。如下图所示:
Kafka 中的每个主题分区都被复制了 n 次,其中的 n 是主题的复制因子(replication factor)。这允许 Kafka 在集群服务器发生故障时自动切换到这些副本,以便在出现故障时消息仍然可用。Kafka 的复制是以分区为粒度的,分区的预写日志被复制到 n 个服务器。 在 n 个副本中,一个副本作为 leader,其他副本成为 followers。顾名思义,producer 只能往 leader 分区上写数据(读也只能从 leader 分区上进行),followers 只按顺序从 leader 上复制日志。
日志复制算法(log replication algorithm)必须提供的基本保证是,如果它告诉客户端消息已被提交,而当前 leader 出现故障,新选出的 leader 也必须具有该消息。在出现故障时,Kafka 会从挂掉 leader 的 ISR 里面选择一个 follower 作为这个分区新的 leader ;换句话说,是因为这个 follower 是跟上 leader 写进度的。
每个分区的 leader 会维护一个 in-sync replica(同步副本列表,又称 ISR)。当 producer 往 broker 发送消息,消息先写入到对应 leader 分区上,然后复制到这个分区的所有副本中。只有将消息成功复制到所有同步副本(ISR)后,这条消息才算被提交。由于消息复制延迟受到最慢同步副本的限制,因此快速检测慢副本并将其从 ISR 中删除非常重要。 Kafka 复制协议的细节会有些细微差别。
副本在什么情况下才算跟上 leader
一个副本如果它没有跟上 leader 的日志进度,那么它可能会被标记为不同步的副本。我通过一个例子来解释跟上(caught up)的含义。假设我们有一个名为 foo 的主题,并且只有一个分区,同时复制因子为 3。假设此分区的副本分别在 brokers 1,2和3上,并且我们已经在主题 foo 上提交了3条消息。brokers 1上的副本是 leader,副本2和3是 followers,所有副本都是 ISR 的一部分。假设 replica.lag.max.messages
设置为4,这意味着只要 follower 落后 leader 的消息不超过3条,它就不会从 ISR 中删除。我们把 replica.lag.time.max.ms
设置为500毫秒,这意味着只要 follower 每隔500毫秒或更早地向 leader 发送一个 fetch 请求,它们就不会被标记为死亡并且不会从 ISR 中删除。
现在假设 producer 往 leader 上发送下一条消息,与此同时,broker 3 上发生了 GC 停顿,现在每个 broker 上的分区情况如下所示:
由于 broker 3 在 ISR中,因此在将 broker 3从 ISR 中移除或 broker 3 上的分区跟上 leader 的日志结束偏移之前,最新消息都是不认为被提交的。注意,由于 border 3 落后 leader 的消息比 replica.lag.max.messages = 4 要小,因此不符合从 ISR 中删除的条件。这意味着 broker 3 上的分区需要从 leader 上同步 offset 为 3 的消息,如果它做到了,那么这个副本就是跟上 leader 的。假设 broker 3 在 100ms 内 GC 完成了,并且跟上了 leader 的日志结束偏移,那么最新的情况如下图:
操作系统优化
大部分Linux发布版本默认的内核参数配置能让大部分应用工作的相当好。但对于实际的Kafka broker场景来说,做稍些改变会提升broker性能。主要涉及的配置:虚拟内存、网络和磁盘挂载(用来存储log segment),一般在 /etc/sysctl.conf
(CentOS系统)。
Virtual Memory
一般来说,Linux的虚拟内存会根据系统负载自动调整。内存页(page)swap到磁盘会显著的影响Kafka的性能,并且Kafka重度使用page cache,如果VM系统swap到磁盘,那说明没有足够的内存来分配page cache。
避免swap的一种方式是设置swap空间为0。但是,swap会在系统崩溃时提供安全机制,或者会在out of memory的情况下阻止操作系统 kill 掉进程。由于这个原因,推荐 vm.swappiness
参数设置为一个非常低的值:1 。这个参数表示 VM系统中的多少百分比用来作为swap空间。
另外一种方式是通过内核调节“脏页”(注:“脏页”会被刷到磁盘上)。Kafka依赖磁盘I/O性能来提高producer的响应时间。这也是为什么通常优先把log segment功能放在可以快速响应的磁盘中(比如,SSD)。这样使得flush进程把“脏数据”写入磁盘前,“脏页”数目就减少了,可以设置 vm.dirty_background_ratio
(表示占用系统内存的百分比)参数的值为 10 以下。大部分应用场景下,vm.dirty_background_ratio
设置为 5 就够用了,要注意了:这个参数值不能设置为 0 ,因为设置为 0 后会引起内核持续刷“脏页”,使得内核的buffer write功能没法施展。
“脏页”的总量可以通过 vm.dirty_ratio
来改变,默认值是 20 (此处也是百分比),这个值的设置范围较大,一般建议设置 60 到 80 为合理的值。但是 vm.dirty_ratio
参数也引来了不小的风险,会造成大量unflush的数据在硬刷到磁盘时产生较长的I/O停顿。如果vm.dirty_ratio 值设置的较大时,强烈建议Kafka开启备份功能,以备系统崩溃。
在设置了这些参数后,需要监控Kafka集群运行时“脏页”的数量,当前“脏页”数量可由如下方式查看(/proc/vmstat
文件):
# cat /proc/vmstat | egrep "dirty|writeback" nr_dirty 3875
nr_writeback 29
nr_writeback_temp 0
|
磁盘
除了考虑磁盘硬件本身和RAID配置外,磁盘的filesystem对Kafka集群的影响最大。虽然有许多filesystem,但最常用的是EXT4或者XFS。在这里XFS文件系统比EXT4稍好,具体原因Google下。
另外一点是,建议开启mount的noatime mount选项。文件系统在文件被访问、创建、修改等的时候会记录文件的一些时间戳,比如:文件创建时间(ctime)、最近一次修改时间(mtime)和最近一次访问时间(atime)。默认情况下,atime的更新会有一次读操作,这会产生大量的磁盘读写,然而atime对Kafka完全没用。
网络
Linux发布版本的网络参数对高网络流量不适用。对于Kafka集群,推荐更改每个socket发送和接收buffer的最大内存:net.core.wmem_default
和 net.core.rmem_default
为128 kb,net.core.wmem_max
和 net.core.rmem_max
为 2 Mb。另外一个socket参数是TCP socket的发送和接收buffer: net.ipv4.tcp_wmem
和 net.ipv4.tcp_rmem
。