kafka学习笔记总结

迷南。 2022-05-27 12:45 272阅读 0赞

### kafka学习笔记总结 ###

参考：  
[http://orchome.com/kafka/index][http_orchome.com_kafka_index]  
[https://blog.csdn.net/qq\_24084925/article/details/78842844][https_blog.csdn.net_qq_24084925_article_details_78842844]

#### 角色说明 ####

Message
        通信的基本单位，每个 producer 可以向一个 topic（主题）发布一些消息。
    Producer
        消息生产者，是消息的产生的源头，负责生成消息并发送到Kafka 服务器上。
    Consumer
        消息消费者，是消息的使用方，负责消费Kafka服务器上的消息。
    Consumer group：
    　　high-level consumer API 中，每个 consumer 都属于一个 consumer group，每条消息只能被 consumer group 中的一个 Consumer 消费，但可以被多个 consumer group 消费。
    Topic
        主题，由用户定义并配置在Kafka服务器，用于建立生产者和消息者之间的订阅关系：生产者发送消息到指定的Topic下，消息者从这个Topic下消费消息。
    Broker
        即Kafka的服务器，用户存储消息，Kafa集群中的一台或多台服务器统称为 broker
    Group
        消费者分组，用于归组同类消费者，在Kafka中，多个消费者可以共同消费一个Topic下的消息，每个消费者消费其中的部分消息，这些消费者就组成了一个分组，拥有同一个分组名称，通常也被称为消费者集群。
    Offset
        消息存储在Kafka的Broker上，消费者拉取消息数据的过程中需要知道消息在文件中的偏移量，这个偏移量就是所谓的Offset。
    partition：
        kafka 分配的单位是 partition。
    　　partition 是物理上的概念，每个 topic 包含一个或多个 partition。
        一个 topic可以分为多个 partition，每个 partition 是一个有序的队列。
        partition中的每条消息都会被分配一个有序的 id（offset）
    replica：
    　　partition 的副本，保障 partition 的高可用。
    leader：
    　　replica 中的一个角色， producer 和 consumer 只跟 leader 交互。
        Leader 是负责给定分区的所有读取和写入的节点。 
        每个分区都有一个服务器充当Leader
    follower：
    　　replica 中的一个角色，从 leader 中复制数据。
    controller：
    　　kafka 集群中的其中一个服务器，用来进行 leader election 以及 各种 failover。
    zookeeper：
    　　kafka 通过 zookeeper 来存储集群的 meta 信息。

#### 概念 ####

我们认为，一个流处理平台具有三个关键能力：

发布和订阅消息（流），在这方面，它类似于一个消息队列或企业消息系统；
    以容错的方式存储消息（流）；
    在消息流发生时处理它们；

kafka应用于2大类应用：

构建实时的流数据管道，可靠地获取系统和应用程序之间的数据。
    构建实时流的应用程序，对数据流进行转换或反应

一些基本概念（The high-level consumer API）：

1.  kafka作为一个集群运行在一个或多个服务器上
2.  kafka集群存储的消息是以topic为类别记录的，不同的topic之间是相互独立的
3.  Message在Broker中通过Log追加的方式进行持久化存储，并进行分区（patitions)
4.  为了减少磁盘写入的次数,broker会将消息暂时buffer起来,当消息的个数(或尺寸)达到一定阀值时,再flush到磁盘,这样减少了磁盘IO调用的次数
5.  无状态导致消息的删除成为难题（可能删除的消息正在被订阅），kafka采用基于时间的SLA(服务水平保证)，消息保存一定时间后会被删除
6.  消息订阅者可以rewind back到任意位置重新进行消费，当订阅者故障时，可以选择最小的offset(id)进行重新读取消费消息
7.  每个topic可以分成几个不同的partition(每个topic有几个partition是在创建topic时指定的)，每个partition存储一部分Message。
8.  partition中的每条Message包含了以下三个属性：
    
    offset 即：消息唯一标识:对应类型：long  
    MessageSize 对应类型：int32  
    data 是message的具体内容。
9.  kafka 的分配单位是 patition，每个 consumer 都属于一个 group，一个 partition 只能被同一个 group 内的一个 consumer 所消费；  
    保障了一个消息只能被 group 内的一个 consuemr 所消费，但是多个 group 可以同时消费这个 partition；
10. 一个consumer可以消费多个partitions中的消息（消费者数据小于Partions的数量时），consumer 采用 pull 模式从 broker 中读取数据；
11. kafka的设计原理决定，对于一个topic，同一个group中不能有多于partitions个数的consumer同时消费，否则将意味着某些consumer将无法得到消息；
12. 一个Topic可以认为是一类消息，每个topic将被分成多partition(区),每个partition在存储层面是append log文件；
13. partition是以文件的形式存储在文件系统中的，任何发布到此partition的消息都会被直接追加到log文件的尾部 - 属于顺序写磁盘（顺序写磁盘效率比随机写内存要高，保障 kafka 吞吐率）
14. partition的数据文件索引基于稀疏存储，每隔一定字节的数据建立一条索引；
15. 分区中的消息都被分了一个序列号，称之为偏移量(offset)，在每个分区中此偏移量都是唯一的（Kafka 保证一个 Partition 内的消息的有序性）；

通讯过程：

客户端打开与服务器端的Socket
    往Socket写入一个int32的数字(数字表示这次发送的Request有多少字节)
    服务器端先读出一个int32的整数从而获取这次Request的大小
    然后读取对应字节数的数据从而得到Request的具体内容
    服务器端处理了请求后，也用同样的方式来发送响应。

kafka有四个核心API：

应用程序使用 Producer API 发布消息到1个或多个topic（主题）。
    应用程序使用 Consumer API 来订阅一个或多个topic，并处理产生的消息。
    应用程序使用 Streams API 充当一个流处理器，从1个或多个topic消费输入流，并生产一个输出流到1个或多个输出topic，有效地将输入流转换到输出流。
    Connector API允许构建或运行可重复使用的生产者或消费者，将topic连接到现有的应用程序或数据系统。例如，一个关系数据库的连接器可捕获每一个变化。

![897247-20161012133845281-1066342716.png][]

#### 工作流程 ####

消费模式：

The high-level consumer API

high-level consumer API 提供了 consumer group 的语义；
    一个消息只能被 group 内的一个 consumer 所消费，且 consumer 消费消息时不关注 offset，最后一个 offset 由 zookeeper 保存。

The SimpleConsumer API

如果你想要对 patition 有更多的控制权，那就应该使用 SimpleConsumer API，比如：
        1. 多次读取一个消息
        2. 只消费一个 patition 中的部分消息
        3. 使用事务来保证一个消息仅被消费一次
    但是使用此 API 时，partition、offset、broker、leader 等对你不再透明，需要自己去管理。你需要做大量的额外工作：
        1. 必须在应用程序中跟踪 offset，从而确定下一条应该消费哪条消息
        2. 应用程序需要通过程序获知每个 Partition 的 leader 是谁
        3. 需要处理 leader 的变更
    使用 SimpleConsumer API 的一般流程如下：
        1. 查找到一个“活着”的 broker，并且找出每个 partition 的 leader
        2. 找出每个 partition 的 follower
        3. 定义好请求，该请求应该能描述应用程序需要哪些数据
        4. fetch 数据
        5. 识别 leader 的变化，并对之作出必要的响应

发布-订阅消息的工作流程（The high-level consumer API）

生产者定期向主题发送消息。
        Kafka代理存储为该特定主题配置的分区中的所有消息。 它确保消息在分区之间平等共享。 如果生产者发送两个消息并且有两个分区，Kafka将在第一分区中存储一个消息，在第二分区中存储第二个消息。
        消费者订阅特定主题。
        一旦消费者订阅主题，Kafka将向消费者提供主题的当前偏移，并且还将偏移保存在Zookeeper系综中。
        消费者将定期请求Kafka新消息。
        一旦Kafka收到来自生产者的消息，它将这些消息转发给消费者。
        消费者将收到消息并进行处理。
        一旦消息被处理，消费者将向Kafka代理发送确认。
        一旦Kafka收到确认，它将偏移更改为新值，并在Zookeeper中更新它。 由于偏移在Zookeeper中维护，消费者可以正确地读取下一封邮件。
        以上流程将重复，直到消费者停止请求。
        消费者可以随时回退/跳到所需的主题偏移量，并阅读所有后续消息。

订阅具有相同 Group ID 的主题的消费者被认为是单个组，并且消息在它们之间共享。 让我们检查这个系统的实际工作流程：

队列消息/用户组的工作流（The high-level consumer API）

生产者以固定间隔向某个主题发送消息。
        Kafka存储在为该特定主题配置的分区中的所有消息，类似于前面的方案。
        单个消费者订阅特定主题，假设 Topic ID为Topic-01，Group ID 为 Group-1 。
        Kafka以与发布 - 订阅消息相同的方式与消费者交互，直到新消费者以相同的组ID 订阅相同主题 Topic-01  1 。
        一旦新消费者到达，Kafka将其操作切换到共享模式，并在两个消费者之间共享数据。 此共享将继续，直到用户数达到为该特定主题配置的分区数。
        一旦消费者的数量超过分区的数量，新消费者将不会接收任何进一步的消息，直到现有消费者取消订阅。 出现这种情况是因为Kafka中的每个消费者将被分配至少一个分区（一个分区同时最多分配给一个消费者），并且一旦所有分区被分配给现有消费者，新消费者将必须等待。
        此功能也称为消费者组。

ZooKeeper的作用

Apache Kafka的一个关键依赖是Apache Zookeeper，它是一个分布式配置和同步服务。 Zookeeper是Kafka代理和消费者之间的协调接口。 Kafka服务器通过Zookeeper集群共享信息。 Kafka在Zookeeper中存储基本元数据，例如关于主题，代理，消费者偏移(队列读取器)等的信息。
        由于所有关键信息存储在Zookeeper中，并且它通常在其整体上复制此数据，因此Kafka代理/ Zookeeper的故障不会影响Kafka集群的状态。 Kafka将恢复状态，一旦Zookeeper重新启动。 这为Kafka带来了零停机时间。 Kafka代理之间的领导者选举也通过使用Zookeeper在领导者失败的情况下完成。

![这里写图片描述][897247-20161012133618468-2107696384.png]

#### Topic创建、删除流程 ####

//创建  
1. controller 在 ZooKeeper 的 /brokers/topics 节点上注册 watcher，当 topic 被创建，则 controller 会通过 watch 得到该 topic 的 partition/replica 分配。  
2. controller从 /brokers/ids 读取当前所有可用的 broker 列表，对于 set\_p 中的每一个 partition：  
2.1 从分配给该 partition 的所有 replica（称为AR）中任选一个可用的 broker 作为新的 leader，并将AR设置为新的 ISR  
2.2 将新的 leader 和 ISR 写入 /brokers/topics/\[topic\]/partitions/\[partition\]/state  
3. controller 通过 RPC 向相关的 broker 发送 LeaderAndISRRequest

//删除  
1. controller 在 zooKeeper 的 /brokers/topics 节点上注册 watcher，当 topic 被删除，则 controller 会通过 watch 得到该 topic 的 partition/replica 分配。  
2. 若 delete.topic.enable=false，结束；否则 controller 注册在 /admin/delete\_topics 上的 watch 被 fire，controller 通过回调向对应的 broker 发送 StopReplicaRequest。

#### 写入流程 ####

1.  producer 先从 zookeeper 的 “/brokers/…/state” 节点找到该 partition 的 leader
2.  producer 将消息发送给该 leader
3.  leader 将消息写入本地 log
4.  followers 从 leader pull 消息，写入本地 log 后 leader 发送 ACK
5.  leader 收到所有 ISR 中的 replica 的 ACK 后，增加 HW（high watermark，最后 commit 的 offset） 并向 producer 发送 ACK

#### 数据传输的事务定义 ####

1.  At most once 消息可能会丢，但绝不会重复传输
2.  At least one 消息绝不会丢，但可能会重复传输
3.  Exactly once 每条消息肯定会被传输一次且仅传输一次

当 producer 向 broker 发送消息时，一旦这条消息被 commit，由于 replication 的存在，它就不会丢；  
但是如果 producer 发送数据给 broker 后，遇到网络问题而造成通信中断，那 Producer 就无法判断该条消息是否已经 commit；  
虽然 Kafka 无法确定网络故障期间发生了什么，但是 producer 可以生成一种类似于主键的东西，发生故障时幂等性的重试多次，这样就做到了 Exactly once，但目前还并未实现；  
所以目前默认情况下一条消息从 producer 到 broker 是确保了 At least once，可通过设置 producer 异步发送实现At most once。

#### kafka-php使用 ####

参考：  
[https://github.com/weiboad/kafka-php/blob/master/README\_CH.md][https_github.com_weiboad_kafka-php_blob_master_README_CH.md]  
[https://github.com/weiboad/kafka-php/blob/master/docs/ch/Configure.md][https_github.com_weiboad_kafka-php_blob_master_docs_ch_Configure.md]

使用 Composer 安装

添加 composer 依赖 nmred/kafka-php 到项目的 composer.json 文件中即可，如：
    
        {
            "require": {
                "nmred/kafka-php": "0.2.*"
            }
        }

Produce:

异步回调方式调用

<?php require '../vendor/autoload.php'; date_default_timezone_set('PRC'); use Monolog\Logger; use Monolog\Handler\StdoutHandler; // Create the logger $logger = new Logger('my_logger'); // Now add some handlers $logger->pushHandler(new StdoutHandler()); $config = \Kafka\ProducerConfig::getInstance(); $config->setMetadataRefreshIntervalMs(10000); $config->setMetadataBrokerList('10.13.4.159:9192'); $config->setBrokerVersion('0.9.0.1'); $config->setRequiredAck(1); $config->setIsAsyn(false); $config->setProduceInterval(500); $producer = new \Kafka\Producer(function() { return array( array( 'topic' => 'test', 'value' => 'test....message.', 'key' => 'testkey', ), ); }); $producer->setLogger($logger); $producer->success(function($result) { var_dump($result); }); $producer->error(function($errorCode) { var_dump($errorCode); }); $producer->send(true);

同步方式调用

<?php
        require '../vendor/autoload.php';
        date_default_timezone_set('PRC');
        use Monolog\Logger;
        use Monolog\Handler\StdoutHandler;
        // Create the logger
        $logger = new Logger('my_logger');
        // Now add some handlers
        $logger->pushHandler(new StdoutHandler());
    
        $config = \Kafka\ProducerConfig::getInstance();
        $config->setMetadataRefreshIntervalMs(10000);
        $config->setMetadataBrokerList('127.0.0.1:9192');
        $config->setBrokerVersion('0.9.0.1');
        $config->setRequiredAck(1);
        $config->setIsAsyn(false);
        $config->setProduceInterval(500);
        $producer = new \Kafka\Producer();
        $producer->setLogger($logger);
    
        for($i = 0; $i < 100; $i++) {
                $result = $producer->send(array(
                        array(
                                'topic' => 'test1',
                                'value' => 'test1....message.',
                                'key' => '',
                        ),
                ));
                var_dump($result);
        }

Consumer

<?php
        require '../vendor/autoload.php';
        date_default_timezone_set('PRC');
        use Monolog\Logger;
        use Monolog\Handler\StdoutHandler;
        // Create the logger
        $logger = new Logger('my_logger');
        // Now add some handlers
        $logger->pushHandler(new StdoutHandler());
    
        $config = \Kafka\ConsumerConfig::getInstance();
        $config->setMetadataRefreshIntervalMs(10000);
        $config->setMetadataBrokerList('10.13.4.159:9192');
        $config->setGroupId('test');
        $config->setBrokerVersion('0.9.0.1');
        $config->setTopics(array('test'));
        //$config->setOffsetReset('earliest');
        $consumer = new \Kafka\Consumer();
        $consumer->setLogger($logger);
        $consumer->start(function($topic, $part, $message) {
            var_dump($message);
        });

[http_orchome.com_kafka_index]: http://orchome.com/kafka/index
[https_blog.csdn.net_qq_24084925_article_details_78842844]: https://blog.csdn.net/qq_24084925/article/details/78842844
[897247-20161012133845281-1066342716.png]: /images/20220527/1f1a148b95344533a0560b29a548b10a.png
[897247-20161012133618468-2107696384.png]: /images/20220527/8f341c592d7a4647872febc2ec560551.png
[https_github.com_weiboad_kafka-php_blob_master_README_CH.md]: https://github.com/weiboad/kafka-php/blob/master/README_CH.md
[https_github.com_weiboad_kafka-php_blob_master_docs_ch_Configure.md]: https://github.com/weiboad/kafka-php/blob/master/docs/ch/Configure.md