kafka全教程

﹏ヽ暗。殇╰゛Y 2022-11-28 13:19 234阅读 0赞

# 1、kafka命令 #

sh  kafka-server-start.sh   -daemon  ../config/server.properties

sh  kafka-server-stop.sh

sh kafka-topics.sh --zookeeper localhost:2181 --list

\#副本数不能大于broker数

sh kafka-topics.sh --zookeeper localhost:2181 --create --topic demo --partitions 10 --replication-factor 3

\#如果不指定topic，则查询所有主题

sh kafka-topics.sh --zookeeper localhost:2191 --describe  --topic  demo

\#修改分区数量，只能增加分区数量，不能减少

sh kafka-topics.sh --zookeeper localhost:2181 --alter --topic demo --partitions 20

\#删除主题，首先标记为已删除状态，等到该主题没有数据交换才会真正删除

sh kafka-topics.sh --zookeeper localhost:2191 --delete --topic my\_t1

sh kafka-consumer-groups.sh --bootstrap-server 10.1.235.54:9081,10.1.235.54:9092,10.1.235.54:9083 --list

\#或者--group指定具体的group

sh kafka-consumer-groups.sh --bootstrap-server 10.1.235.54:9081,10.1.235.54:9082,10.1.235.54:9083 --describe --all-groups

\#生产者

sh kafkaserver-1/bin/kafka-console-producer.sh --broker-list 10.1.235.54:9081,10.1.235.54:9082,10.1.235.54:9083 --topic topic826 --producer.config kafkaserver-1/config/producer.properties

\#消费者

sh kafkaserver-1/bin/kafka-console-consumer.sh --bootstrap-server 10.1.235.54:9081,10.1.235.54:9082,10.1.235.54:9083 --topic topic826 --from-beginning --consumer.config kafkaserver-1/config/consumer.properties

\#查看日志文件

sh kafkaserver-1/bin/kafka-run-class.sh kafka.tools.DumpLogSegments --files kafkaserver-1/kafka-logs/topic826-0/00000000000000000000.log --print-data-log

\#查看偏移量索引文件

sh kafkaserver-1/bin/kafka-dump-log.sh --files  kafkaserver-1/kafka-logs/topic826-0/00000000000000000000.index

\#查看时间戳索引文件

sh kafkaserver-1/bin/kafka-dump-log.sh --files  kafkaserver-1/kafka-logs/topic826-0/00000000000000000000.timeindex

# 2、kafka简介 #

kafka是一个高吞吐量、分布式的消息队列，现在是apache项目的一部分

kafka核心模块使用scala开发，支持多语言客户端

目前越来越多的分布式处理系统可以与kafka进行集成，如flume、apache storm、spark、flink等

## 2.1、kafka背景 ##

![2020082617141413.png][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hsXzE4MDM_size_16_color_FFFFFF_t_70][]

## 2.2、kafka基本概念 ##

### 2.2.1、主题 ###

### 2.2.2、消息 ###

![2020082617480062.png][]

### 2.2.3、分区和副本 ###

每个主题分成一个或多个分区（Partition）,每个分区由一系列有序、不可变的消息组成，是一个有序队列

分区在物理上对应一个文件夹，命名规则：$\{topicName\}-分区编号

分区使得kafka在并发处理上变得更加容易，理论上分区越多吞吐量越高，但是这要根据集群实际环境和业务场景而定

分区是kafka保证消息被顺序消费以及对消息进行负载均衡的基础

kafka只能保证一个分区之内消息的有序性，跨分区消息不能保证有序性，每条消息被追加到相应的分区（顺序写入磁盘），这是kafka高吞吐量的一个重要保证

与传统消息队列不同的是，kafka不会立即删除已消费的消息，由于磁盘的限制也不会一直被存储（kafka提供相应的清理旧数据的策略）

### 2.2.4、leader副本和follower副本 ###

如果不设置leader副本，那么为了保证数据一致性，需要开通n\*n条通路来同步数据

引入leader副本之后，leader副本负责读写，follower副本从leader副本同步数据，且只负责读，这样只需要开通n-1条通路来同步数据即可

如果leader失效，则应该通过选举算法重新选出新的leader副本

### 2.2.5、偏移量 ###

旧版消费者将消费偏移量信息保存在zookeeper中，新版消费者保存在kafka内部的一个主题中

任何发布到分区的消息会追加到日志文件中，而每条消息在日志文件中的位置都会对应一个递增的偏移量，偏移量是一个逻辑值，并不表示在磁盘上的物理位置

kafka几乎不允许对消息进行随机读写，但可以指定消费的起始偏移量

### 2.2.6、消费者和消费者组 ###

单播：同一个消费者组内只会有一个消费者消费到消息

广播：不同消费者组的消费者可以消费到同一条消息

### 2.2.7、ISR ###

kafka在zookeeper中动态维护了一个ISR(in-sync Replia)，即保存同步的副本列表，该列表保存的是与Leader副本保持消息同步的所有副本对应的代理节点id

如果一个Follower副本宕机（比如代理被关闭、心跳检测过期、网络延迟、进程崩溃、落后太多），则该follower副本节点将从ISR列表中移除

### 2.2.8、日志段 ###

### 2.2.9、代理 ###

### 2.2.10、生产者 ###

### 2.2.11、zookeeper ###

kafka利用zookeeper保存相应的元数据信息：

代理节点信息、集群信息旧版消费者信息及其偏移量信息、主题信息、分区状态信息、分区副本分配方案信息、动态配置信息等

kafka在启动或运行过程中会在zookeeper上创建相应节点来保存元数据信息，kafka通过监听机制在这些节点上注册监听器会监听数据的变化，从而由zookeeper负责管理维护kafka

## 2.3、kafka设计概述 ##

### 2.3.1、kafka设计动机 ###

设计初衷：实时处理大规模数据的平台

kafka与其他消息队列相比更像是一个数据库日志系统

### 2.3.2、kafka特性 ###

**a）消息持久化**

*  kafka依赖文件系统（磁盘）存储消息
 *  大家普遍认为磁盘速度远小于内存，但是磁盘的快慢取决于我们如何去应用磁盘
 *  线性（顺序）读写磁盘的速度 > 内存随机读写
 *  并且现代操作系统提供了预读（read-ahead）和延迟写（write-behind）技术
 *  kafka是基于JVM的，如果采用内存存储，java对象的内存消耗非常高，且随着java对象的增加GC也会越来越频繁
 *  传统的消息系统数据持久化一般会提供一个类似于B树的随机访问数据结构来维护消息的元数据，B树的时间复杂度为O(log n),B树虽然有一定的优势，但会随着数据的增长性能会下降。而kafka的数据持久化是顺序追加，时间复杂度为O(1)，性能和数据大小没有关系

**b）高吞吐量**

*  充分利用磁盘的顺序读写特性
 *  kafka在数据写入和数据同步采用零拷贝技术，采用sendFile()函数，sendFile()是在两个文件描述符之间直接传递数据，完全在内核中操作
 *  支持数据压缩和批量发送
 *  一个主题又分为多个分区
 *  kafka每秒支持数百万级别的消息

**c）集群扩展**

kafka依赖于zookeeper对集群进行协调管理，可以将多台廉价的PC服务器搭建成一个大规模的消息系统，水平扩展时无需将整个集群停机，集群能够自动感知，重新进行负载均衡及数据复制

**d）多客户端支持**

**e）与主流大数据框架集成**

Flume、Hadoop、HBase、Hive、Spark、Storm

**f）Kafka Streams**

kafka在0.10版本后引入，Kafka Streams是一个用java语言实现的用于流处理的jar文件

**g）安全性**

*  通过SSL和SASL(kerberos)，SASL/PLAIN验证机制支持生产者、消费者与代理连接时的身份验证
 *  支持代理与zookeeper连接身份验证
 *  通信时数据加密
 *  客户端读写权限认证
 *  kafka支持与其他外部认证授权服务的集成

**h）支持数据备份(副本机制)**

**i）数据压缩**

支持Gzip、snappy、LZ4这3种压缩方式，通常把多条消息放在一起组成MessageSet，然后把MessageSet放到一条消息里面去

**j）轻量级**

### 2.3.3、应用场景 ###

*  消息队列
 *  应用监控
 *  网站用户行为跟踪
 *  流处理
 *  持久化日志

# 3、Kafka核心组件 #

## 3.1、日志管理器LogManager ##

### 3.1.1、作用 ###

*  日志的创建与删除
 *  日志检索
 *  日志加载和恢复
 *  检查点
 *  日志文件刷写磁盘
 *  日志清理

### 3.1.2、日志结构 ###

日志文件大小由log.segment.bytes指定，默认为1G,同时kafka提供了根据时间来切片的机制，即若数据文件没有达到log.segment.bytes指定的阈值，但达到了log.roll.ms或log.roll.hours（还是log.segment.ms?）指定的阈值，同样会创建新的日志段

**消息存储结构：**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hsXzE4MDM_size_16_color_FFFFFF_t_70 1][]

时间戳类型为0时，表示创建时间，为1表示消息追加时间

在实际存储一条消息时，还包括12字节的额外开销，8字节用于表示消息的偏移量，4字节表示消息总长度

### 3.1.3、偏移量索引文件 ###

数据文件分段使得可以在一个较小的数据文件中查找对应offset的Message了，但是这依然需要顺序扫描才能找到对应offset的Message。为了进一步提高查找的效率，Kafka为每个分段后的数据文件建立了索引文件，文件名与数据文件的名字是一样的，只是文件扩展名为.index  
索引文件中包含若干个索引条目，每个条目表示数据文件中一条Message的索引。索引包含两个部分（均为4个字节的数字），分别为相对offset和position。

*  相对offset：因为数据文件分段以后，每个数据文件的起始offset不为0，相对offset表示这条Message相对于其所属数据文件中最小的offset的大小。举例，分段后的一个数据文件的offset是从20开始，那么offset为25的Message在index文件中的相对offset就是25-20 = 5。存储相对offset可以减小索引文件占用的空间。
 *  position，表示该条Message在数据文件中的绝对位置。只要打开文件并移动文件指针到这个position就可以读取对应的Message了。

index文件中并没有为数据文件中的每条Message建立索引，而是采用了稀疏存储的方式，每隔一定字节的数据建立一条索引。这样避免了索引文件占用过多的空间，从而可以将索引文件保留在内存中。但缺点是没有建立索引的Message也不能一次定位到其在数据文件的位置，从而需要做一次顺序扫描，但是这次顺序扫描的范围就很小了

在Kafka中，索引文件的实现类为OffsetIndex，它的类图如下：

![20150121162054898][]

主要的方法有：

*  append方法，添加一对offset和position到index文件中，这里的offset将会被转成相对的offset。
 *  lookup, 用二分查找的方式去查找小于或等于给定offset的最大的那个offset

**如何查找Message**

![search][]

比如：要查找绝对offset为7的Message：

1.  首先是用二分查找确定它是在哪个LogSegment中，自然是在第一个Segment中
2.  打开这个Segment的index文件，也是用二分查找找到offset小于或者等于指定offset的索引条目中最大的那个offset。自然offset为6的那个索引是我们要找的，通过索引文件我们知道offset为6的Message在数据文件中的位置为9807
3.  打开数据文件，从位置为9807的那个地方开始顺序扫描直到找到offset为7的那条Message

### 3.2.4、时间戳索引文件 ###

待补充

### 3.2.5、日志清理 ###

日志清理策略（不管是哪种清理策略，只针对不活跃的日志段）：

*  日志删除delete
 *  日志压缩compact：日志不会被删除，而是进行去重，这种模式要求record必须有key，去重时只保留最新的那个key

通过如下配置指定：

*  log.cleanup.policy 代理级别

**a）日志删除**

日志管理器启动后，会启动一个后台定时任务线程用于定时删除日志段文件，每隔$\{log.retention.check.interval.ms\}毫秒检测一下是否该删除日志段文件，默认是5分钟

基于日志保留时长

*  log.retention.hours
 *  log.retention.minutes
 *  log.retention.ms

基于日志大小

*  log.retention.bytes

**b）日志压缩**

待补充

# 4、kafka topic级别配置 #

配置topic级别参数时，相同(参数)属性topic级别会覆盖全局的，否则默认为全局配置属性值。

创建topic参数可以设置一个或多个--config "Property(属性)",下面是创建一个topic名称为"my-topic"例子，它设置了2个参数max message size 和 flush rate.

## 4.1、创建topic时配置参数 ##

bin/kafka-topics.sh --zookeeper 192.168.2.225:2183/config/mobile/mq/mafka02 --create --topic my-topic --partitions 1   --replication-factor 1 
    --config max.message.bytes=64000 --config flush.messages=1

## 4.2、修改topic时配置参数 ##

覆盖已经有topic参数，下面例子修改"my-topic"的max message属性

bin/kafka-topics.sh --zookeeper192.168.2.225:2183/config/mobile/mq/mafka02  --alter --topic my-topic  --config max.message.bytes=128000

## 4.3、删除topic级别配置参数 ##

bin/kafka-topics.sh --zookeeper192.168.2.225:2183/config/mobile/mq/mafka02  --alter --topic my-topic  --delete-config max.message.bytes

## 4.4、Topic级别配置属性表 ##

<table> 
 <tbody> 
 <tr> 
 <td style="vertical-align:top;"> Property(属性) </td> 
 <td style="vertical-align:top;"> Default(默认值) </td> 
 <td style="vertical-align:top;"> Server Default Property(server.properties) </td> 
 <td style="vertical-align:top;"> 说明(解释) </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;"> cleanup.policy </td> 
 <td style="vertical-align:top;"> delete </td> 
 <td style="vertical-align:top;"> log.cleanup.policy </td> 
 <td style="vertical-align:top;"> 日志清理策略选择有：delete和compact主要针对过期数据的处理，或是日志文件达到限制的额度，会被&nbsp; topic创建时的指定参数覆盖 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;"> delete.retention.ms </td> 
 <td style="vertical-align:top;"> 86400000 (24 hours) </td> 
 <td style="vertical-align:top;"> log.cleaner.delete.retention.ms </td> 
 <td style="vertical-align:top;"> 对于压缩的日志保留的最长时间，也是客户端消费消息的最长时间，同log.retention.minutes的区别在于一个控制未压缩数据，一个控制压缩后的数据。会被topic创建时的指定参数覆盖 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;"> flush.messages </td> 
 <td style="vertical-align:top;"> None </td> 
 <td style="vertical-align:top;"> log.flush.interval.messages </td> 
 <td style="vertical-align:top;"> log文件”sync”到磁盘之前累积的消息条数,因为磁盘IO操作是一个慢操作,但又是一个”数据可靠性"的必要手段,所以此参数的设置,需要在"数据可靠性"与"性能"之间做必要的权衡.如果此值过大,将会导致每次"fsync"的时间较长(IO阻塞),如果此值过小,将会导致"fsync"的次数较多,这也意味着整体的client请求有一定的延迟.物理server故障,将会导致没有fsync的消息丢失. </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;"> flush.ms </td> 
 <td style="vertical-align:top;"> None </td> 
 <td style="vertical-align:top;"> log.flush.interval.ms </td> 
 <td style="vertical-align:top;"> 仅仅通过interval来控制消息的磁盘写入时机,是不足的.此参数用于控制"fsync"的时间间隔,如果消息量始终没有达到阀值,但是离上一次磁盘同步的时间间隔达到阀值,也将触发. </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;"> index.interval.bytes </td> 
 <td style="vertical-align:top;"> 4096 </td> 
 <td style="vertical-align:top;"> log.index.interval.bytes </td> 
 <td style="vertical-align:top;"> 当执行一个fetch操作后，需要一定的空间来扫描最近的offset大小，设置越大，代表扫描速度越快，但是也更好内存，一般情况下不需要搭理这个参数 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;"> message.max.bytes </td> 
 <td style="vertical-align:top;"> 1,000,000 </td> 
 <td style="vertical-align:top;"> message.max.bytes </td> 
 <td style="vertical-align:top;"> 表示消息的最大大小，单位是字节 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;"> min.cleanable.dirty.ratio </td> 
 <td style="vertical-align:top;"> 0.5 </td> 
 <td style="vertical-align:top;"> log.cleaner.min.cleanable.ratio </td> 
 <td style="vertical-align:top;"> 日志清理的频率控制，越大意味着更高效的清理，同时会存在一些空间上的浪费，会被topic创建时的指定参数覆盖 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;"> retention.bytes </td> 
 <td style="vertical-align:top;"> None </td> 
 <td style="vertical-align:top;"> log.retention.bytes </td> 
 <td style="vertical-align:top;"> topic每个分区的最大文件大小，一个topic的大小限制&nbsp; =&nbsp;分区数*log.retention.bytes。-1没有大小限log.retention.bytes和log.retention.minutes任意一个达到要求，都会执行删除，会被topic创建时的指定参数覆盖 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;"> retention.ms </td> 
 <td style="vertical-align:top;"> None </td> 
 <td style="vertical-align:top;"> log.retention.minutes </td> 
 <td style="vertical-align:top;"> 数据存储的最大时间超过这个时间会根据log.cleanup.policy设置的策略处理数据，也就是消费端能够多久去消费数据 log.retention.bytes和log.retention.minutes达到要求，都会执行删除，会被topic创建时的指定参数覆盖 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;"> segment.bytes </td> 
 <td style="vertical-align:top;"> 1 GB </td> 
 <td style="vertical-align:top;"> log.segment.bytes </td> 
 <td style="vertical-align:top;"> topic的分区是以一堆segment文件存储的，这个控制每个segment的大小，会被topic创建时的指定参数覆盖 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;"> segment.index.bytes </td> 
 <td style="vertical-align:top;"> 10 MB </td> 
 <td style="vertical-align:top;"> log.index.size.max.bytes </td> 
 <td style="vertical-align:top;"> 对于segment日志的索引文件大小限制，会被topic创建时的指定参数覆盖 </td> 
 </tr> 
 <tr> 
 <td style="vertical-align:top;"> log.roll.hours </td> 
 <td style="vertical-align:top;"> 7 days </td> 
 <td style="vertical-align:top;"> log.roll.hours </td> 
 <td style="vertical-align:top;"> 这个参数会在日志segment没有达到log.segment.bytes设置的大小，也会强制新建一个segment会被&nbsp; topic创建时的指定参数覆盖 </td> 
 </tr> 
 </tbody> 
</table>

# 5、kafka生产者配置 #

kafka为生产者提供三种消息确认机制（acks）,通过request.required.acks来配置：

acks=0 生产者不用等待代理返回确认信息

acks=1 生产者需要等待Leader副本成功将消息写入日志文件中（仍有数据丢失的风险：如果Follower还没来得及同步，而此时Leader机器宕机了）

acks=-1 生产者需要等待Leader副本和ISR列表中的所有副本都将消息写入到日志文件中

# 6、kafka架构 #

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hsXzE4MDM_size_16_color_FFFFFF_t_70 2][]

[2020082617141413.png]: /images/20221124/8c7ce10171d74cc7991acf2a8a0cdb95.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hsXzE4MDM_size_16_color_FFFFFF_t_70]: /images/20221124/f7efb4550f974c2c8e4adeaaa79e4b48.png
[2020082617480062.png]: /images/20221124/1b62a7313c34408c8343552c3f0abd2b.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hsXzE4MDM_size_16_color_FFFFFF_t_70 1]: /images/20221124/bafb4dd3cbc848db99c47bd98e0d19e6.png
[20150121162054898]: /images/20221124/cd9d7e288ac84cabad6ed6f97164069d.png
[search]: /images/20221124/5671cce5837540429b9b491fc56783da.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3hsXzE4MDM_size_16_color_FFFFFF_t_70 2]: /images/20221124/1e00f3927e4840cda88016fca331d12f.png