专栏首页后台全栈之路入门 Kafka 你所需要了解的基本概念和开发模式
原创

入门 Kafka 你所需要了解的基本概念和开发模式

  团队在日常工作中,一般情况下使用的消息队列是腾讯云 CKafka。CKafka 提供了高可靠的开箱即用消息队列能力,让我们在日常能够放心使用,减少花在运维上的投入。不过即便如此,我们还是需要学习 Kafka 的一些基本概念和功能,从而在实际应用中嗯能够充分高效、高质量地利用 Kafka 的能力。


业务基本概念

本小节主要说明的是在软件业务层面,我们使用 Kafka 中会接触到的概念

消息 Message

  对于一个消息队列系统,最基础的自然是 “消息”。在 Kafka 中,“消息” 就是 Message,是在 Kafka 中数据传输的基本单位。多个消息会被分批次写入 Kafka 中。这同一批次的消息,就称为一消息。

生产者消费者

  生产者和消费者的概念就很好理解了:产生消息的服务就称为 “生产者” Producer,也称为 “发布者” PublisherWriter

  而需要获取消息的服务就称为 “消费者” Consumer,也称为 “订阅者” SubScriberReader

主题 Topic 和 分区 Partition

  在 Kafka 中,所有的消息并不是还有一条队列。Kafka 的消息通过 Topic 进行分类。而一个 Topic 可以被区分为多个 “分区” Partitions

  这里需要注意的是,在 同一个 partition 内部,消息的顺序是能够保证的。也就是说:如果消息A到达 partition 的时间早于消息B,那么消费者在获取消息的时候,必然是先获得消息A之后,才能够获取到消息B。

  但是,如果在多个 partitions 之间,消息的顺序就无法保证了。比如当消费者监听多个 partitions 时的话,消息A和消息B被读取出来的时间无法保证。

  那这么一来,partition 有什么用呢?实际上 Partition 是用来做负载均衡的。当 comsumer 将消息发到一个 topic 上时,Kafka 默认会将消息尽量均衡地分发到多个 partitions 上。作为消费者监听 topic 时,需要配置监听哪些 partitions。一个 consumer 可以监听多个 partitions,comsumer 和 partition(s) 的对应关系也称为 “所有权关系”

偏移量 Offset

  Offset 是一个递增的整数值,由 Kafka 自动递增后自动写入每一个 partition 中。在同一个 partition 中,一个 offset 值唯一对应着一条 message。此外,由于 offset 是递增的,因此也可以用来区分多个 message 之间的顺序。Consumer 的重启动作并不影响 offset 的值,因为这是 Kafka 来进行维护的数值。

Broker 和 集群

  一个独立的 Kafka server 就称为一个 broker。一个或多个 broker 可以组成一个 “集群” broker cluster。Kafka 虽然是一个分布式的消息队列系统,但是在集群中,Kafka 依然是准中心化的系统架构。也就是说每一个集群中依然是有一台主 broker,称为 controller

  每一个 cluster 会自动选举一个 cluster controller 出来,controller 需要负责以下操作:

  1. 管理 cluster
  2. 将 partition 分配给 broker 和监控 broker。

  在 cluster 中,一个 partition 会从属于一个 broker,这个 broker 也会称作该 partition 的 leader。同时该 partition 也可以分配给多个 broker,进行 分区复制——如果其中一个 broker 失效了,那么其余的 broker 可以尽快接管 leader 的位置。如果是使用云原生的 Kafka,我们一般就不需要太担心这个问题。


安装/运维基本概念

Kafka 部署架构

  如果是运维自己安装 Kafka 的话,需要提前安装的软件是 JavaZookeeper。我当时就非常疑惑怎么多了一个 Zookeeper?实际上 Kafka 是使用 Zookeeper 来保存 cluster 中的元数据和消费者信息。这里体现出了 Java 强大和完善的生态圈,各种方案都能够找到已有的轮子。

  Zookeeper 也支持集群部署。Zookeeper 集群称为 “群组” Ensemble。因为 ensemble 也是使用了选举机制,因此每个 ensemble 中有奇数个节点,不建议超过7个。如果我们使用了云原生的 Kafka,就不需要过多关心这个细节啦。

Topic 参数

  部署好了 broker 和 Zookeeper 之后,我们就可以创建 topics 了。创建 topic 时有一些参数需要进行配置。主要的有以下几项需要特别留意:

  • num.partitions: 新 topic 默认的分区数。在后续运维中,partition 的数量只会增加,不会减少。在腾讯云 CKafka 中,这对应着 “分区数” 配置
  • log.retension.ms: 按照时间决定 topic 中的数据可以保存多久。这对应 CKafka 界面中的 “retension.ms” 参数
  • log.retension.bytes: 按照存储空间决定 topic 中的数据可以保存多少。该参数在 CKafka 中不支持
  • log.segment.bytes: 表示按照存储空间决定日志片段文件的大小。该参数在 CKafka 中不支持
  • log.segment.ms: 表示按照时长决定日志片段大小。对应 CKafka 界面中的 “segment.ms” 参数。不是必要参数

如何选择 Partitions 的数量?

  前面提到,在同一个 partition 中,消息的顺序是能够得到保证的。因此对于一个小型的、对可靠性要求不高、但是对顺序性要求很高的系统而言,或许可以使用单 partition 的方案。

  但是这个方案其实是非常危险的:

  • 首先,单一 partition 就意味着 consumer 也只能有一个,否则会出现消息重复消费的问题。在一个生产项目中进行单点部署,这几乎是不可接受的
  • 虽然在 Kafka 内部,单一 partition 内的消息顺序能够得到保证,但如果生产者未能得到保证的话,那么 kafka 内的消息顺序依然不是真实的。因此对于有强顺序要求的消息队列系统中,不建议使用时间顺序,而是采用逻辑顺序/逻辑时钟来区分消息的先后。

  因此在实际生产环境中,我们应当适当地分配 partition 的数量。如果对顺序性有要求,那么不应该依赖 kafka 的顺序机制,而是使用额外的机制来保证。


Kafka 生产者

架构图

  生产者向 Kafka broker 发送消息一般是用各语言的 SDK 来完成的。下面框图中是 SDK 完成的逻辑。首先 producer 在发送 message 之前,需要将 message 封装到 producer record 中,record 包含的必填信息是 topic 和 value(也就是 message 正文)信息。此外还可选 partition 和 key 信息,不过相对少用。Key 参数的作用后文会作说明。

  当消息被写入 Kafka broker 之后,broker 会回调到 SDK 中,将消息最终落地的 partition 和 partition 中的 offset 信息返回给 SDK,并最终视需要返回给 Producer。

消息发送

  Kafka 生产者有两种消息发送方式:同步异步

  同步发送方式就是生产者发出的每一个消息,都需要按照上面的结构图的流程处理:消息发出后等待 Kafka broker 的结果响应之后再做进一步的处理。Kafka broker 返回的错误中包含了两种错误:

  1. 可重试错误: 当遇到这一类错误时,生产者可以直接重新尝试发送。比如网络错误、集群错误等等。
  2. 不可重试错误: 当遇到这一类错误时,生产者只能考虑告警、记录、修改软件逻辑等等。比如消息过大等等。

  异步发送方式就是生产者通过 SDK 发送消息之后就直接返回;SDK 在后台处理消息的发送、响应处理,然后通过回调告知生产者以进行进一步的处理。

生产者参数

  生产者启动之前也有一些参数可进行配置。读者可以在各语言的 SDK 中具体查找:

  • acks: 消息发送给 Kafka broker,由于实际上会有多个 broker,因此消息是需要复制多份的。该参数表示需要等待多少个 broker 的响应,才视为消息发送成功:
    • 0: 表示不需要等待 broker 响应
    • 1: 表示 leader 响应即可
    • all: 表示需要所有的 broker 响应
  • buffer.memory: 生产者的缓冲区大小。单位是 message 的数量。当缓冲区满了之后,SDK 会根据 maxblock.ms 等待并阻塞一段时间之后再进行重试。如果缓冲区还是满了的状态,则会抛出异常或返回错误
  • compression.type: 消息压缩格式,可选值为: snappy, gzip, lz4
  • retries: 重试次数,重试间隔为 retry.backoff.ms,默认是 100ms
  • batch.size: 一个批次的数据大小,字节数。为了减少网络传输中的消耗,Kafka 生产者并不是一个消息就通过一次发送发出去,而是组成一个个批次进行发送。当一个批次的大小达到这个参数时,则会马上发出。
  • linger.ms: 一个批次发送之前的缓冲时间。当批次的尺寸未达到 batch.size 的话,SDK 也不会一直按住 message 不发送,而是等待一段时间之后也会把内存中的批次发出
  • client id: 自定义字符串,用于标识生产者
  • max.id.flight.requests.per.connection: 这个参数指的是收到服务器响应之前,生产者可以发送的消息数。设置为 1 可以保证消息顺序,但是相应的效率就下降了
  • request.timeout.ms: 生产者发送数据之后等待响应的时间

Key 的作用

  在 producer record 中的 key 有两个用途:

  1. 作为消息的附加消息
  2. 可以用来决定写入到哪一个分区。默认分区器可以使拥有相同 key 的消息写入同一个分区。
  3. 如果 key == null,则默认采用轮询方式写入分区
  4. 如果 key 非空,则根据哈希结果决定分区

  生产者也可以通过自定义分区器来实现业务的具体分区功能,具体参见各语言的 SDK


Kafka 消费者

  一个 Kafka 的消费者是从属于其对应的 comsumer group 的,每一个 group 订阅一个 topic,每个 consumer 消费一部分的消息。整个 group 内部通过消费不同的 partition 实现负载均衡。每一个 group 都有一个 group.id 用于标识一个消费者群组,这在业务中就对应着一个消费者业务。

  不要让消费者数量多于分区数量,否则会导致出现重复消费的问题。因此在 partition 选用时,宜多不宜省。更多的分区数量也能够更加合理地分配 consumer 之间的负载。

分区再均衡 Partitions Reoke / Rebalance

  每个消费者可以对应一个或多个 partition;多个 consumer 组成一个 group,覆盖 topic 的全部 partitions。但是当 consumer 和 partitions 数量发生变化时,需要重新分配所有权关系。这个动作就称为 Rebalance。至于是热切换还是冷切换,则由业务方决定。

  消费者在调用 subscribe() 监听消息时,可以传入一个 ConsumerRebalanceListener 实例来监听事件。其中需要关注的事件有:

  • onPartitionsRevoked(): 这是再平衡开始之前的事件。注意此时消费者应停止消费,并且 commit 已完成但尚未 commit 的 offset 值
  • onPartitionsAssigned(): 这是再平衡结束,也就是重新分配分区结束之后的时间。大部分情况下消费者也不需要特别处理什么,不过可以在这里进行一些消费过程的重启动作

Commit 和 Offset

  前文提到,一个 message 能够与 kafka 中的一个 partition 中的一个 offset 值一一对应。对于消费者而言,partition-offset 对也可以用于标识当前 comsumer 已经获取到的消息的进度,也可以用于消费者在 kafka 中进行历史消息的寻址。

  当对某个 message 消费完成后,消费者会将 offset 值提交到 kafka 中,从而让 Kafka 识别并保存某个 comsumer group 的消费进度。下一次 consumer 再请求事件时,默认会从该 offset 往后继续获取。Consummer 向 Kafka 更新 offset 的这一动作就称为 “提交” commit

  如果 consumer 发生崩溃,或者有新的 consumer 加入 group,就会触发 rebalance。完成 rebalancing 之后,每个 consumer 有可能会被分配到不同的分区。为了能够继续之前的工作,consumer 需要读取每一个分区最后一次提交的 offset,然后从指定的 offset 继续处理。这个操作,一般在 SDK 中就完成了。但是在上述切换过程中,由于分布式系统的分布式、异步特性,我们不可避免的还是可能遇到一些不一致的情况,具体表现为消息的重复处理和漏处理。所以我们在任何时候都不能简单依赖 Kafka 本身提供的消息队列机制,而是在各自的业务系统中也需要进行一定的防御式编程,避免错误处理出现。

  一般而言,SDK 有下面几种 commit 方式:

  • 自动提交: enable.auto.committrue 时,API 定时、异步地进行 commit。因此,如果在触发了再均衡的时候还有部分数据未 commit,那么在再均衡之后在其他的消费者中就有可能发生重复消费
  • 主动提交: enable.auto.commitfalse 时,业务方需要主动调用相关 API 进行 commit。
  • (主动的)异步提交: 其实就是主动提交的异步版,简单而言就是开一个后台异步 commit 的过程。
  • 提交特定的 offset: 这种模式就是显式地 commit 具体 partition 的某个 offset 值。

参考资料


本文章采用 知识共享署名-非商业性使用-相同方式共享 4.0 国际许可协议 进行许可。

原作者: amc,欢迎转载,但需要按照以上许可协议注明出处。

原文标题:入门 Kafka 你所需要了解的基本概念和开发模式

发布日期:2020-12-01

原文链接:https://cloud.tencent.com/developer/article/1755160

原创声明,本文系作者授权云+社区发表,未经许可,不得转载。

如有侵权,请联系 yunjia_community@tencent.com 删除。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • 腾讯 Tars 基础框架手动搭建——填掉官方 Guide 的坑

    腾讯 Tars 是腾讯内部使用的 TAF(Tencent Application Framework)的对外开源版,去掉了许多冗杂多余的部分。该框架集开发、运维...

    amc
  • 一张图清晰解释微信三方平台获取授权流程

    微信公众平台体系,大家最为熟悉的,一个是公众号,另一个就是小程序。如果需要使用公众号的高级功能,那么大家首先想到的就是自组开发团队来干这些活儿。绝大多数情况下,...

    amc
  • 腾讯 Tars-Go 服务获取自定义模版(配置)值

    腾讯 Tars 框架中,有两种可以称之为 “配置” 的地方:其中一个是可以自定义的,在 Tars 管理页面中称为 “服务配置”。在这里,可以按照开发者喜欢的格式...

    amc
  • Kafka系列文章第1篇之Kafka是什么

    如果有幸目睹过系统从零到一的演变过程,大家估计都会有一种感叹,就是随着业务复杂度和流量的不断上升,系统变得越来越难以维护,面对高额的维护成本,攻城师们不得不对现...

    z小赵
  • Kafka的生成者、消费者、broker的基本概念

    kafka是一款基于发布与订阅的消息系统。它一般被称为“分布式提交日志”或者“分布式流平台”。文件系统或者数据库提交日志用来提供所有事物的持久化...

    chenchenchen
  • 图文详解:Kafka到底有哪些秘密让我对它情有独钟呢?

    最近很少发文,一是开始总结了一下自己做了两个多月的公号了,都收获了什么,学到了什么。

    浅羽技术
  • Kafka 简介

    在Kafka中,客户端和服务器之间的通信是通过一种简单的,高性能的,语言不可知的TCP协议完成的。

    小忽悠
  • Kafka集群消息积压问题及处理策略

    通常情况下,企业中会采取轮询或者随机的方式,通过Kafka的producer向Kafka集群生产数据,来尽可能保证Kafk分区之间的数据是均匀分布的。

    大数据学习与分享
  • Kafka:高吞吐量、消息精确一次语义以及保证消息顺序

    Kafka 最初由 Linkedin 公司开发,是一个分布式、支持分区的、多副本的,基于 Zookeeper 协调的分布式消息系统,其最大的特性就是可以实时的处...

    CG国斌
  • Kafka 面试题,看这一篇就够了

    原文链接:https://url.cn/5h7oDHS

    业余草

扫码关注云+社区

领取腾讯云代金券

,,