Kakfa 起初是由 LinkedIn 公司开发的一个分布式的消息系统,后成为 Apache 的一部分,它使用 Scala 编写,以可水平扩展和高吞吐率而被广泛使用。目前越来越多的开源分布式处理系统如 Cloudera、Apache Storm、Spark 等都支持与 Kafka 集成。

体系架构

生产者使用 push 向 broker 发送消息,消费者使用 pull 模式拉取并消费消息

  • producer 可能是服务器日志,业务数据
  • broker 消息中间件节点,支持水平扩展,一般 broker 数量越多,集群吞吐率越高
  • consumer
  • zookeeper 集群, Kafka 依赖 zookeeper 管理

存储策略

Kafka 中消息是以 topic 进行分类的,生产者通过 topic 向 Kafka broker 发送消息,消费者通过 topic 读取数据。topic 在物理层面以 partition 为分组,partition 还可以细分为 segment。

假设 Kafka 消息文件存储目录

log.dirs=/tmp/kafka-logs

假设 partition 数量为 4

/bin/kafka-topics.sh –create –zookeeper localhost:2181 –partitions 4 –topic mytopic –replication-factor 4

然后就能在 /tmp/kafka-logs 目录中看到 4 个目录。

在 Kafka 文件存储中,同一个 topic 下有多个不同的 partition,每个 partiton 为一个目录,partition 的名称规则为:topic 名称 + 有序序号,第一个序号从 0 开始计,partition 是实际物理上的概念,而 topic 是逻辑上的概念。

每个 partition(目录)相当于一个巨型文件被平均分配到多个大小相等的 segment(段)数据文件中(每个 segment 文件中消息数量不一定相等)这种特性也方便 old segment 的删除,即方便已被消费的消息的清理,提高磁盘的利用率。每个 partition 只需要支持顺序读写就行,segment 的文件生命周期由服务端配置参数(log.segment.bytes,log.roll.{ms,hours}等若干参数)决定。

segment 文件由两部分组成,分别为.index文件和.log文件,分别表示为 segment 索引文件和数据文件。这两个文件的命令规则为:partition 全局的第一个 segment 从 0 开始,后续每个 segment 文件名为上一个 segment 文件最后一条消息的 offset 值,数值大小为 64 位,20 位数字字符长度,没有数字用 0 填充。.index 索引文件存储大量的元数据,.log 数据文件存储大量的消息,索引文件中的元数据指向对应数据文件中 message 的物理偏移地址。

每条消息都具有固定的物理结构,包括:offset(8 Bytes)、消息体的大小(4 Bytes)、crc32(4 Bytes)、magic(1 Byte)、attributes(1 Byte)、key length(4 Bytes)、key(K Bytes)、payload(N Bytes) 等等字段,

topic vs partition

一类 topic 可以认为一类消息,每个 topic 会被分成多个 partition,每个 partition 使用 append log 文件存储。

在创建 topic 时可以在 $KAFKA_HOME/config/server.properties 中指定 partition 的数量,也可以在 topic 创建之后去修改 partition 的数量。

# The default number of log partitions per topic. More partitions allow greater
# parallelism for consumption, but this will also result in more files across
# the brokers.
num.partitions=4

broker 重要配置

Kafka 有非常多的参数可以控制其工作和运行,大部分情况下默认值就可以,但是要用到生产上,就需要自定义一些参数来适应不同的环境。broker 中有一些非常重要的参数,这里列举一些,关于 broker 所有其他的参数可以参考 Kafka 的官方文档。

broker.id

每个 kafka broker 都需要有一个整型的唯一标识,这个标识通过 broker.id 来设置。默认的情况下,这个数字是 0, 但是它可以设置成任何值。需要注意的是,需要保证集群中这个 id 是唯一的。这个值是可以任意填写的,并且可以在必要的时候从 broker 集群中删除。比较好的做法是使用主机名相关的标识来做为 id, 比如,你的主机名当中有数字相关的信息,如 hosts1.example.com,host2.example.com, 那么这个数字就可以用来作为 broker.id 的值。

port

默认启动 kafka 时,监听的是 TCP 的 9092 端口,端口号可以被任意修改。如果端口号设置为小于 1024, 那么 kafka 需要以 root 身份启动。但是并不推荐以 root 身份启动。

zookeeper.connect

这个参数指定了 Zookeeper 所在的地址,它存储了 broker 的元信息。在前一章节的例子中,Zookeeper 是运行在本机的 2181 端口上,因此这个值被设置成 localhost:2181。这个值可以通过分号设置多个值,每个值的格式都是 hostname:port/path, 其中每个部分的含义如下:

  • hostname 是 zookeeper 服务器的主机名或者 ip 地址
  • port 是服务器监听连接的端口号
  • /path 是 kafka 在 zookeeper 上的根目录。如果缺省,会使用根目录

log.dirs

这个参数用于配置 Kafka 保存数据的位置,Kafka 中所有的消息都会存在这个目录下。可以通过逗号来指定多个目录,kafka 会根据最少被使用的原则选择目录分配新的 parition。注意 kafka 在分配 parition 的时候选择的规则不是按照磁盘的空间大小来定的,而是分配的 parition 的个数多小。

num.recovery.thread.per.data.dir

kafka 可以配置一个线程池,线程池的使用场景如下:

  • 当正常启动的时候,开启每个 parition 的文档块 segment
  • 当失败后重启时,检查 parition 的文档块
  • 当关闭 kafka 的时候,清除关闭文档块

默认,每个目录只有一个线程。最好是设置多个线程数,这样在服务器启动或者关闭的时候,都可以并行的进行操作。尤其是当非正常停机后,重启时,如果有大量的分区数,那么启动 broker 将会花费大量的时间。注意,这个参数是针对每个目录的。比如,num.recovery.threads.per.data.dir 设置为 8, 如果有 3 个 log.dirs 路径,那么一共会有 24 个线程。

auto.create.topics.enable

在下面场景中,按照默认的配置,如果还没有创建 topic,kafka 会在 broker 上自动创建 topic:

  • 当 producer 向一个 topic 中写入消息时
  • 当 cosumer 开始从某个 topic 中读取数据时
  • 当任何的客户端请求某个 topic 的信息时

在很多场景下,这都会引发莫名其妙的问题。尤其是没有什么办法判断某个 topic 是否存在,因为任何请求都会创建该 topic。如果你想严格的控制 topic 的创建,那么可以设置 auto.create.topics.enable 为 false。

num.partitions

这个参数用于配置新创建的 topic 有多少个分区,默认是 1 个。注意 partition 的个数只可以被增加,不能被减少。这就意味着如果想要减少主题的分区数,那么就需要重新创建 topic。

Kafka 通过分区来对 topic 进行扩展,因此需要使用分区的个数来做负载均衡,如果新增了 broker, 那么就会引发重新负载分配。这并不意味着所有的主题的分区数都需要大于 broker 的数量,因为 kafka 是支持多个主题的,其他的主题会使用其余的 broker。需要注意的是,如果消息的吞吐量很高,那么可以通过设置一个比较大的分区数,来分摊压力。

log.retention.ms

用于配置 kafka 中消息保存的时间,也可以使用 log.retention.hours, 默认这个参数是 168 个小时,即一周。另外,还支持 log.retention.minutes 和 log.retention.ms。这三个参数都会控制删除过期数据的时间,推荐还是使用 log.retention.ms。如果多个同时设置,那么会选择最小的那个。

过期时间是通过每个 log 文件的最后修改时间来定的。在正常的集群操作中,这个时间其实就是 log 段文件关闭的时间,它代表了最后一条消息进入这个文件的时间。然而,如果通过管理员工具,在 brokers 之间移动了分区,那么这个时候会被刷新,就不准确了。这就会导致本该过期删除的文件,被继续保留了。

log.retention.bytes

这个参数也是用来配置消息过期的,它会应用到每个分区,比如,你有一个主题,有 8 个分区,并且设置了 log.retention.bytes 为 1G, 那么这个主题总共可以保留 8G 的数据。注意,所有的过期配置都会应用到 patition 粒度,而不是主题粒度。这也意味着,如果增加了主题的分区数,那么主题所能保留的数据也就随之增加。

log.segment.bytes

用来控制 log 段文件的大小,而不是消息的大小。在 kafka 中,所有的消息都会进入 broker, 然后以追加的方式追加到分区当前最新的 segment 段文件中。一旦这个段文件到达了 log.segment.bytes 设置的大小,比如默认的 1G, 这个段文件就会被关闭,然后创建一个新的。一旦这个文件被关闭,就可以理 解成这个文件已经过期了。这个参数设置的越小,那么关闭文件创建文件的操作就会越频繁,这样也会造成大量的磁盘读写的开销。

log.segment.ms

控制段文件关闭的时间,它定义了经过多长时间段文件会被关闭。

message.max.bytes

这个参数用于限制生产者消息的大小,默认是 1000000, 也就是 1M。生产者在发送消息给 broker 的时候,如果出错,会尝试重发;但是如果是因为大小的原因,那生产者是不会重发的。另外,broker 上的消息可以进行压缩,这个参数可以使压缩后的大小,这样能多存储很多消息。需要注意的是,允许发送更大的消息会对性能有很大影响。更大的消息,就意味着 broker 在处理网络连接的时候需要更长的时间,它也会增加磁盘的写操作压力,影响 IO 吞吐量。

Kafka 特点

  • kafka 也被设计为多个消费者去读取任意的单个消息流而不相互影响;同时多个 kafka 消费者也可以选择作为一个组的一部分,来分担一个消息流,确保这整个组,这个消息只被消费一次
  • 基于硬盘的消息保存,消息将按照持久化配置规则存储在硬盘上。这个可以根据每个 topic 进行设置,允许根据不同的消费者的需求不同设置不同消息流的保存时间不同,持久化保存意味着一旦消费者来不及消费或者突然出现流量高峰,而不会有丢失数据的风险。同样也意味着消息可以由 consumer 来负责管理,比如消费消息掉线了一段时间,不需要担心消息会在 producer 上累积或者消 息丢失,consumer 能够从上次停止的地方继续消费
  • 水平扩展能力强,扩展可以在集群正常运行的时候进行,对于整个系统的运作没有影响,集群如果要同时容忍更多的故障的话,可以配置更高的 replication factors
  • 高性能

Kafka 的使用场景

关于该部分,Kafka 官方的入门教程中有非常详细的介绍,包括从最早在 LinkedIn 中为记录用户访问数据设计该系统,到后面日志,消息处理,到流处理等等,Kafka 有着非常广阔的使用场景。

更多关于如何使用命令行启动 Kafka,还有一些基础内容可以参考翻译的 Kafka 中文文档

reference