kafka流处理平台快速入门

小灰灰 2023-07-14 10:44 20阅读 0赞

一、Kafka简介  
二、kafka基本概念  
三、kafka结构设计  
四、kafka场景与应用  
五、简单使用案例  
六、代码案例  
七、kafka高级特性

--------------------

**一、Kafka简介**  
Kafka流处理平台由LinkedIn开发，2011年被apache收纳。  
**Kafka的三个关键特性：**

1.  发布和订阅记录流，类似于消息队列或企业消息传递系统。
2.  以容错的持久方式存储记录流。
3.  当记录流产生时，就可以进行处理

**Kafka通常被用在两大类应用：**

1.  构建实时流数据管道，在系统和应用程序之间可靠地获取数据
2.  构建实时流应用程序以转换或响应数据流

**Kafka特点**  
**1.分布式：**

*  多分区(partitions)
 *  多副本(replications)\[提供容错\]
 *  多订阅者\[Topic可以有一个或多个订阅者，订阅者数量不大于Partions数量\]
 *  基于Zookeeper调度\[存储Broker、Topic、Partions信息等等\]

**2.高性能：**

*  高吞吐量\[每秒几十万的吞吐量\]，低延迟，高并发，事件复杂度为o(1)

**3.持久性和扩展性：**

*  数据可持久化、容错性、支持在线水平扩展、消息自动平衡

--------------------

**二、kafka基本概念**

<table> 
 <thead> 
  <tr> 
   <th>名称</th> 
   <th>解释</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>Producer</td> 
   <td>消息和数据的生产者，向Kafka的一个topic发布消息的进程/代码/服务</td> 
  </tr> 
  <tr> 
   <td>Cumsumer</td> 
   <td>消息和数据的消费者，订阅数据(Topic)并且处理其发布的消息的进程/代码/服务。</td> 
  </tr> 
  <tr> 
   <td>Consumer Group</td> 
   <td>逻辑概念，对于同一个topic，会广播给不同的group，一个group中，只有一个consumer可以消费该topic。</td> 
  </tr> 
  <tr> 
   <td>Broker</td> 
   <td>物理概念，Kafka集群中的每个Kafka节点</td> 
  </tr> 
  <tr> 
   <td>Topic</td> 
   <td>逻辑概念，Kafka消息的类别，对数据进行区分、隔离</td> 
  </tr> 
  <tr> 
   <td>Partition</td> 
   <td>物理概念，Kafka下数据存储的基本单元。一个Topic数据，会被分散存储到多个Partition，每一个Partition时有序的，kafka会将一个Partition放在一个Broker里面。</td> 
  </tr> 
  <tr> 
   <td>Replication</td> 
   <td>同一个Partition可能会有多个Replica，多个Replica之间数据是一样的。</td> 
  </tr> 
  <tr> 
   <td>Replication Leader</td> 
   <td>一个Partition的多个Replica上，需要一个Leader负责该Partition上与Produce和Consumer交互。</td> 
  </tr> 
  <tr> 
   <td>RepicaManager</td> 
   <td>负责管理当前broker所有分区和副本的信息，处理KafkaController发起的一些请求，副本状态的切换、添加/读取消息、选举Leader等。</td> 
  </tr> 
 </tbody> 
</table>

**名词概念延伸**  
**Partition**  
1.每一个Topic被切分为多个Partitions  
2.消费者数目少于或等于Partition的数目(否作多个消费者消费同一个Partition)  
3.Broker Group中每一个Broker保存Topic的一个或多个Partitions，同一个Partitions不会被多个Broker同时保存，当Partitions很大时，会切分给各个broker存储。  
Consumer Group中仅有一个Consumer读取Topic的一个或多个Partitions并且是唯一的Consumer

**Replication**  
1.当集群中有Broker挂掉的情况，系统可以主动地使用Replicas提供服务。  
2.系统默认设置每一个Topic的replication系数为1(没有副本)，可以在创建Topic时单独设置  
**Replication特点**  
1.基本单位时Topic的Partition;  
2.所有的读和写都从Leader进，Followers只是做为备份  
3.Follwers必须能够及时负责Leader的数据

--------------------

**三、基本结构**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM4OTAwNTY1_size_16_color_FFFFFF_t_70]  
**kafka消息队列运作图**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM4OTAwNTY1_size_16_color_FFFFFF_t_70 1]  
从上图看出，Kafka强依赖于zookeeper，那么其作用是：  
1.Broker信息都存储在Zookeeper上  
2.Topic信息和Partitions分布信息都存储在Zookeeper

**Kafka消息结构**  
![在这里插入图片描述][20200314170941508.PNG]

<table> 
 <thead> 
  <tr> 
   <th>字段</th> 
   <th>解释</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>offset</td> 
   <td>记录当前这个消息它所处于的偏移量</td> 
  </tr> 
  <tr> 
   <td>Length</td> 
   <td>当前消息长度</td> 
  </tr> 
  <tr> 
   <td>CRC32</td> 
   <td>校验信息完整性，避免信息不完整导致数据生产或消费失败</td> 
  </tr> 
  <tr> 
   <td>Magic</td> 
   <td>快速判定该消息是否是kafka消息，否则遗弃</td> 
  </tr> 
  <tr> 
   <td>attributes</td> 
   <td>存储当前消息的属性</td> 
  </tr> 
  <tr> 
   <td>Timestamp</td> 
   <td>时间戳</td> 
  </tr> 
  <tr> 
   <td>Key Length</td> 
   <td>key的长度</td> 
  </tr> 
  <tr> 
   <td>Key</td> 
   <td>key的值，无长度限制</td> 
  </tr> 
  <tr> 
   <td>Value Length</td> 
   <td>value长度</td> 
  </tr> 
  <tr> 
   <td>Value</td> 
   <td>value的值，无长度限制</td> 
  </tr> 
 </tbody> 
</table>

--------------------

**四、Kafka的应用场景及应用**

*  消息队列： 优势在于高吞吐、分区、副本等容错性，大规模消息处理系统；稳定，同时它的消息可被重复消费，内置数据持久化具有更低延迟。媲美ActiveMQ、
 *  行为跟踪：属于发布订阅的扩展应用，当我们需要跟踪用户浏览行为时，可以将所有的浏览记录以发布订阅的方式将数据实时记录到topic里，那这些被订阅者获取后做进一步处理。
 *  元数据监控：通常用于操作记录监控模块，记录操作信息，可理解为运维性质的监控。
 *  日志收集：可让数据进行流动起来，比起传统以文件为中心处理系统例如ELK。
 *  流处理：保存收集上游的流数据对接到下游的计算/存储。
 *  事件源：将状态转移按事件排列的序列，可以回溯事件状态变更。
 *  持久性日志(commit log)系统之外持久性记录，故障恢复的以一种机制。

--------------------

**五、Kafka简单案例(命令行)**  
**解压与安装【本章略过该方式】**  
Zookeeper下载：http://zookeeper.apache.org/releases.html\#download  
Kafka下载：http://kafka.apache.org/downloads  
安装见此篇： https://www.cnblogs.com/wonglu/p/8687488.html  
**Docker启动【本章使用该方式】**

[root@zan71 ~]# docker run -d --name zookeeper -p 2181:2181 -t wurstmeister/zookeeper
    [root@zan71 ~]# docker run --name kafka01 -p 9092:9092 -e KAFKA_BROKER_ID=0 -e KAFKA_ZOOKEEPER_CONNECT=192.168.91.66:2181 -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://192.168.91.66:9092 -e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 -d  wurstmeister/kafka

**简单生产者**  
1.进入kafka容器

[root@zan71 ~]# docker exec -it kafka01 /bin/bash

2.创建主题，副本为1(无副本)，分区为3

bash-4.4# /opt/kafka/bin/kafka-topics.sh --create --zookeeper 192.168.245.128:2181 --replication-factor 1 --partitions 3 --topic test-topic

3.列出当前的主题

bash-4.4# /opt/kafka/bin/kafka-topics.sh --list --zookeeper 192.168.245.128:2181

4.启动生产者交互终端：

bash-4.4# /opt/kafka/bin/kafka-console-producer.sh --broker-list 192.168.245.128:9092 --topic test-topic

**简单消费者**  
5.1：启动从头消费终端(之前消费过的再次消费)\[添加参数–from-beginning\]

bash-4.4# /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test-topic --from-beginning

5.2: 启动消费未被消费的终端(只会消费剩下没被消费过的消息)

bash-4.4# /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test-topic

5.3 测试，往生产者交互终端输入数据，查看消费终端屏幕打印

--------------------

**六、Kafka代码案例**  
Django+Celery+Kafka+Prometheus打造高性能日志处理平台`【敬请等待】`

--------------------

**七、Kafka高级特性**  
**7.1 消息事务**  
为什么要支持事务？

*  满足 "读取-处理-写入” 模式，即数据一致性，类似SQL的comit，比如说你往ATM打钱，直到将金额写入数据库中，才算成功。否则中间只要有错误，将给你退钱；避免出现你存了钱但是没有写进数据库。
 *  流处理需求的不断增强
 *  不准确的数据处理的容忍度在不断降低。提供数据的准确性

**数据传输的事务定义：**

*  最多一次：消息不会被重复发送，最多被传输一次，但也有可能一次不传输。
 *  最少一次：消息不会被漏发送，最少被传输一次，但也有可能被重复传输
 *  精确的一次(Exactly once)：不会漏传输也不会重复传输，每个消息都被传输一次而且仅仅被传输一次，这是大家所期望的

**事务保证：**  
内部重试问题： Procdure幂处理。  
多分区原子写入：要么被写入，要被被丢弃。  
避免僵尸实例：

*  每个事务Producer分配一个transactional.id，在进程重写启动时能够识别相同的Producer实例
 *  Kafka增加了一个与transactional.id相关的epoch，存储每个transactional.id内部元数据。
 *  一旦epoch被触发，任何具有相同的transactional.id和更旧的epoch的Producer被视为僵尸，Kafka会拒绝来自这些Procedure的后续事务性写入  
    **kafka如何实现多分区原子的操作？**  
    原子是读取处理写入的过程，假如topic1的偏移量X处读取消息A，然后它对消息进行处理之后，将这个消息命名为B，并写入到topic2这一整个过程。整个操作是原子的。  
    如何认为读取写入是原子的？成功消费并且发布，或者不发布，那前提是成功消费。  
    如何判断一个消息是成功消费？当读取A时，偏移量会被标记为已消费。  
    何时判断已消费？Kafka会将偏移量写入offsets topic内部的kafka topic来记录offsets commit。所以当偏移量被提交offsets topic时才认定是成功消费

**7.2零拷贝**  
网络传输持久性日志块：例如日常的生产和消费的消息就是日志块，传输这个的时候本身很消耗性能，kafka是如何提供性能？-零拷贝技术。使用Java Nio channel.transforTo()方法即可实现零拷贝数据传输过程，底层是Linux sendfile系统调用。

**零拷贝是怎样提供数据传输性能？**  
**文件传输到网络的公共数据路径：** 把磁盘上的一个日志块发送到网络上，要经历：  
1.操作系统将数据从磁盘读入到内核空间的页缓存(页缓存基于磁盘上第一层缓存，只有读取到该层缓存上，系统才能快速地拷贝到其他地方)  
2.应用程序将数据从内核空间读入到用户空间缓存中(对于Linux来说，内核空间是应用程序无法直接操作的，应用程序能操作的是当前用户空间，因此应用程序需要将数据从内核空间读入到用户空间缓存中)  
3.应用程序将数据写回到内核空间的socket缓存中(因为要发到网络上，所以要发到socket缓存中)  
4.操作系统将数据从socket缓冲区复制到网卡缓冲区，以便将数据经网络发出。  
总结：磁盘->网络需要经4次拷贝

**零拷贝过程**  
1.操作系统将数据从磁盘读入到内核空间的页缓存(该过程是省不掉的，无论如何都需将数据读取到页缓存中)  
2.将数据的位置和长度的信息的描述符增加到内核空间(socker缓冲区)  
3.操作系统通过描述符将数据从内核拷贝到网卡缓冲区，以便数据经网络发出  
总共2次拷贝，零拷贝的意思是内核空间和用户空间的交互拷贝次数为0，所以称之为零拷贝。

**文件传输到网络的公共数据路径演变**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM4OTAwNTY1_size_16_color_FFFFFF_t_70 2]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM4OTAwNTY1_size_16_color_FFFFFF_t_70]: /images/20230529/f03acbc905b74cdaa8460b0fb3db3d31.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM4OTAwNTY1_size_16_color_FFFFFF_t_70 1]: /images/20230529/c8e85ddc4ad64d25990762934588acec.png
[20200314170941508.PNG]: /images/20230529/fb496137396543ca91cdd0d3412a7005.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM4OTAwNTY1_size_16_color_FFFFFF_t_70 2]: /images/20230529/a8e2e258edff46de9b2603ba8b25b593.png