Kafka-架构、命令行、生产消费数据、保存机制、拦截器、kafka Streams

￡神魔★判官ぃ 2024-04-17 19:50 54阅读 0赞

# Kafka #

### 一、消息队列 ###

普通消息传递，由A传递给B，容意发生的问题：

接收端B挂掉，发送端A不知，继续发送数据

发送端A的发送数据速度和接收端B速度必须保持一致

#### 1 消息队列模式 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwOTI5OTIx_size_16_color_FFFFFF_t_70]

添加消息队列后，A发送数据给消息队列，由消息队列选择发送模式发送给B

发送模式主要有2种：

点对点模式，一对一，由接收端主动拉取  
 由接收端B去开启一个实时监控消息队列的线程，来查看消息队列是否有数据，自主控制拉取速度，注意，消息消费完后立即删除，不保留历史消息

发布/订阅者模式，一对多，由消息队列主动推送  
 可以发送给多个客户端，不需要专门的线程来监控

#### 2 消息队列优点 ####

解耦 --A，B可以进行修改，只需保证遵守同样的接口约束

冗余 --消息队列对数据持久化知道数据被完全处理

扩展性 --解耦

灵活性&峰值处理能力

可恢复性 --数据持久化，AB挂掉不会影响已传输的数据

顺序保证性 --先进先出（partition）

缓冲 --均衡两端速度

异步通信 --消息可能不会被及时处理，需要存放一段时间后再处理

### 二、kafka架构 ###

#### 1 简介 ####

**高吞吐的分布式消息系统，默认存储磁盘，默认保存7天**

kafka由scala写成

时间策略删除，kafka和很多消息系统不一样，很多消息系统是消费完了我就把它删掉，而kafka是根据时间策略删除，而不是消费完就删除，在kafka里面没有一个消费完这么个概念，只有过期这样一个概念

#### 2 架构 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwOTI5OTIx_size_16_color_FFFFFF_t_70 1]

producer 消息生产者，两种机制，1基于轮询，2key的Hash，若key为空则只能轮询写

broker 一个独立的kafka服务器就是一个broker

一个集群由多个broker组成，一个broker可以容纳多个topic

组成kafka集群的节点，broker间没有主从关系，依赖于ZK协调

负责消息的读写和存储，每个broker可以管理多个partition

接收来自生产者的消息，为消息设置偏移量，提交消息到磁盘保存

响应消费者请求，对读取分区做出响应，返回已经提交到磁盘上的消息

每个集群有一个broker是集群控制器（自动选举）

控制器将分区分配给broker和监控broker

一个分区属于一个broker，broker是分区的master

一个分区可以分配给多个broker，提供了消息冗余，多个副本之间主从切换

topic 分类 一类消息/消息队列

每个topic是由多个partition组成，可指定数量

partition 做负载均衡

组成topic的单元，直接接触磁盘，消息是append到每个partition上的

每个partition内部消息是强有序的FIFO，每个消息都有一个序号-offset

创建partition可以指定副本

一个partition只对应一个broker

consumer 每个consumer都有自己的消费者组

每个消费者组在同一个topic消费时，该topic中数据只能被消费一次

不同的消费者组消费同一个topic时互不影响

0.8前由zk管理offset 0.8之后由kafka自身管理

CG Consumer Group 含有多个consumer

kafka用来实现一个topic消息的广播(发给所有的consumer)和单播(发给任意一个consumer)的手段

一个topic可以有多个CG

topic的消息会复制(逻辑))到所有的CG，但每个partion只会把消息发给该CG中的一个consumer

如果需要实现广播，只要每个consumer有一个独立的CG就可以了

要实现单播，只要所有的consumer在同一个CG

CG还可以将consumer进行自由的分组而不需要多次发送消息到不同的topic

offset kafka的存储文件都是按照offset.kafka来命名，用offset做名字的好处是方便查找

the first offset就是00000000000.kafka，2049的位置，只要找到2048.kafka的文件即可

zookeeper 存储元数据，broker,topic,consumer…(offset)

producer不在zk中注册，消费者在zk中注册

### 三、kafka命令行操作 ###

提交spark命令
    	Standalone-client
    	./spark-submit --master spark://sxt001:7077 --class 项目包名.obj名 ../examples/jars/jar包名 分配partition数量
    	./spark-submit --master spark://sxt001:7077 --deploy mode client --class 项目包名.obj名 ../examples/jars/jar包名 分配partition数量
    	Standalone-cluster
    	./spark-submit --master spark://sxt001:7077 --deploy mode cluster --class 项目包名.obj名 ../examples/jars/jar包名 分配partition数量
    	Yarn-client
    	将jar包导入spark的examples文件夹中，进入spark/bin，执行命令：
    	./spark-submit --master yarn --class 项目包名.obj名 ../examples/jars/jar包名 分配partition数量
    	./spark-submit --master yarn-client --class 项目包名.obj名 ../examples/jars/jar包名 分配partition数量
    	./spark-submit --master yarn --deploy-mode client  --class 项目包名.obj名 ../examples/jars/jar包名 分配partition数量
    	Yarn-cluster
    	./spark-submit --master yarn-cluster --class 项目包名.obj名 ../examples/jars/jar包名 分配partition数量
    	./spark-submit --master yarn --deploy-mode cluster  --class 项目包名.obj名 ../examples/jars/jar包名 分配partition数量
    
    测试
    分别启动Zookeeper、Kafka、Flume集群。
    	zkServer.sh start
    		bin/kafka-server-start.sh config/server.properties
    		bin/flume-ng agent -n a1 -c conf -f conf/fk.conf -Dflume.root.logger=DEBUG,console
    
    创建topic：（不用）
    		bin/kafka-topics.sh --zookeeper sxt002:2181,sxt003:2181,sxt004:2181 --create --replication-factor 2 --partitions 3 --topic testflume
    		bin/kafka-topics.sh --zookeeper sxt002:2181,sxt003:2181,sxt004:2181 --create --replication-factor 2 --partitions 1 --topic LogError
    
    查看
    		bin/kafka-topics.sh --zookeeper sxt002:2181,sxt003:2181,sxt004:2181 --list
    		
    查看某个Topic的详情
    		bin/kafka-topics.sh sxt002:2181,sxt003:2181,sxt004:2181 --describe --topic first
    
    
    启动消费者：		
    		bin/kafka-console-consumer.sh --zookeeper sxt002:2181,sxt003:2181,sxt004:2181 --from-beginning --topic testflume
    		bin/kafka-console-consumer.sh --bootstrap-server sxt002:9092,sxt003:9092,sxt004:9092 --topic testflume
    		--from-beginning：会把first主题中以往所有的数据都读取出来。根据业务场景选择是否增加该配置。
    
    启动生产者
    		./kafka-console-producer.sh  --topic  topic2017 --broker-list sxt002:9092,sxt003:9092,sxt004:9092
    
    
    启动集群后，在3个虚拟机中的任意一台执行2条命令
    		bin/kafka-topics.sh --zookeeper sxt002:2181,sxt003:2181,sxt004:2181 --create --replication-factor 2 --partitions 1 --topic mylog_cmcc
    		bin/kafka-console-consumer.sh --zookeeper sxt002:2181,sxt003:2181,sxt004:2181 --from-beginning --topic mylog_cmcc
    
    删除	
    		bin/kafka-topics.sh --zookeeper sxt002:2181,sxt003:2181,sxt004:2181 --delete --topic xxxx
    		在日志文件中找到相关topic xxxx删除文件夹	rm -rf ./xxxx/
    		取ZK删除元数据进入bin/zkCli.sh		./zkCli.sh  		ls /
    		ls /brokers/topics		rmr /brokers/topics/xxxx
    		ls /
    		ls /config/topics		rmr /config/topics/xxxx
    		ls /
    		ls /adimn/delete_topics rmr /adimn/delete_topics/xxxx
    
    手动维护offset（以组为单位）
    		启动消费者1(server)，生产数据,查看组，list后会有名称，打开别名描述(可以查看消费到哪个位置)
    		bin/kafka-consumer-groups.sh --bootstrap-server sxt002:9092,sxt003:9092,sxt004:9092 --list
    		bin/kafka-consumer-groups.sh --bootstrap-server sxt002:9092,sxt003:9092,sxt004:9092 --group console-consumer-39716 --describe
    		还可以自增消费组，也可以查看相应信息

### 四、kafka生产数据流程 ###

#### 1 写入方式 ####

producer采用推（push）模式将消息发布到broker，每条消息都被追加（append）到分区（patition）中，属于顺序写磁盘（顺序写磁盘效率比随机写内存要高，保障kafka吞吐率）

#### 2 Partition分区 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwOTI5OTIx_size_16_color_FFFFFF_t_70 2]

每个Partition中的消息都是有序的，生产的消息被不断追加到Partition log上，其中的每一个消息都被赋予了一个唯一的offset值

#### 3 分区原因 ####

（1）方便在集群中扩展，每个Partition可以通过调整以适应它所在的机器，而一个topic又可以有多个Partition组成，因此整个集群就可以适应任意大小的数据了

（2）可以提高并发，因为可以以Partition为单位读写了

#### 4 分区原则 ####

（1）指定了patition，则直接使用；

（2）未指定patition但指定key，通过对key的value进行hash出一个patition,若key为空，则只能轮询

（3）patition和key都未指定，使用轮询选出一个patition

#### 5 分区策略 ####

假如6个broker，主题包含10个分区，复制系数为3，则kafka有30个分区副本。

1、 在broker之间平均分布分区副本，保证每个broker可以分到5个副本

2、 确保每个分区的每个副本分布在不同的broker上。

3、 如果为broker指定了机架信息，则尽可能把每个分区副本分配到不同机架的broker上。

#### 6 副本（Replication） ####

同一个partition可能会有多个replication

没有replication的情况下，一旦broker 宕机，其上所有 patition 的数据都不可被消费，同时producer也不能再将数据存于其上的patition

引入replication之后，同一个partition可能会有多个replication，而这时需要在这些replication之间选出一个leader，producer和consumer只与这个leader交互，其它replication作为follower从leader 中复制数据

#### 7 写入数据 ####

流程：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwOTI5OTIx_size_16_color_FFFFFF_t_70 3]

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

**Tips:**

如何保证kafka数据不会被丢失 —调节ACK机制的参数（0,1,all）另ACK检查机制为all

**Tips:**

如果想要重复消费一个Topic数据：

1 重新新建一个消费者组  
 需要在代码中添加参数：(ConsumerConfig.AUTO\_OFFSET\_RESET\_CONFIG,“earliest”)  
 2 低级API

### 五、kafka保存消息 ###

高吞吐的分布式消息系统，默认存储磁盘，默认保存7天

不经过内存直接存入磁盘(0拷贝)

#### 1 存储方式 ####

物理上把topic分成一个或多个patition（对应 server.properties 中的num.partitions=3配置），每个patition物理上对应一个文件夹（该文件夹存储该patition的所有消息和索引文件）

#### 2 存储策略 ####

无论消息是否被消费，kafka都会保留所有消息。有两种策略可以删除旧数据：

1）基于时间：log.retention.hours=168

2）基于大小：log.retention.bytes=1073741824

需要注意的是，因为Kafka读取特定消息的时间复杂度为O(1)，即与文件大小无关，所以这里删除过期文件与提高 Kafka 性能无关

#### 3 ZooKeeper ####

主要保存一些元数据信息

#### 4 文件格式 ####

kafka消息和偏移量保存在文件里。磁盘上的数据格式与从生产者发送过来的以及发送给消费者的消息格式是一样的

#### 5 索引 ####

消费者可以从kafka的任意可用偏移量位置开始读取消息。kafka为每个分区维护了一个索引。索引把偏移量映射到片段文件和偏移量在文件里的位置

索引也被分成片段，在删除消息时也删除相应的索引。kafka不维护索引的校验和。如果索引出现损坏，kafka会通过重新读取消息并记录偏移量和位置来重新生成索引

#### 6 清理 ####

kafka会在包含脏记录的主题数量达到50%时进行清理，避免太过频繁的清理，避免存在太多脏记录

### 六、kafka消费数据流程 ###

kafka提供了两套consumer API：高级ConsumerAPI 和 低级Consumer API。

#### 1 高级API ####

1）高级API优点 --简单，自动管理

高级API 写起来简单

不需要自行去管理offset，系统通过zookeeper自行管理

不需要管理分区，副本等情况，系统自动管理

消费者断线会自动根据上一次记录在zookeeper中的offset去接着获取数据（默认设置1分钟更新一下zookeeper中存的offset）

可以使用group来区分对同一个topic 的不同程序访问分离开来（不同的group记录不同的offset，这样不同程序读取同一个topic才不会因为offset互相影响）

2）高级API缺点 --无法手动设置

不能自行控制offset（对于某些特殊需求来说）

不能细化控制如分区、副本、zk等

#### 2 低级API ####

1）低级 API 优点 --自己控制

能够让开发者自己控制offset，想从哪里读取就从哪里读取。

自行控制连接分区，对分区自定义进行负载均衡

对zookeeper的依赖性降低

如：offset不一定非要靠zk存储，自行存储offset即可，比如存在文件或者内存中

2）低级API缺点 --复杂

太过复杂，需要自行控制offset，连接哪个分区，找到分区leader 等。

#### 3 消费者组 ####

一个消费者可以消费多个分区

同一个消费者组中的不同消费者不能消费同一份数据

#### 4 消费方式 ####

consumer采用pull（拉）模式从broker中读取数据

push（推）模式很难适应消费速率不同的消费者，因为消息发送速率是由broker决定的。它的目标是尽可能以最快速度传递消息，但是这样很容易造成consumer来不及处理消息，典型的表现就是拒绝服务以及网络拥塞。而pull模式则可以根据consumer的消费能力以适当的速率消费消息

pull模式缺点：

如果kafka没有数据，消费者可能会陷入循环中，一直等待数据到达。为了避免这种情况，我们在我们的拉请求中有参数，允许消费者请求在等待数据到达的“长轮询”中进行阻塞，并且可选地等待到给定的字节数，以确保大的传输大小

### 七、kafka拦截器(interceptor) ###

#### 1 原理 ####

在原有数据的基础上添加一些简单的逻辑，达到过滤数据的效果

Producer拦截器(interceptor)是在Kafka 0.10版本被引入的，主要用于实现clients端的定制化控制逻辑

producer允许用户指定多个interceptor按序作用于同一条消息从而形成一个拦截链(interceptor chain)

Intercetpor的实现接口是org.apache.kafka.clients.producer.ProducerInterceptor，

#### 2 接口方法 ####

其定义的方法包括：

（1）configure(configs)

获取配置信息和初始化数据时调用

（2）onSend(ProducerRecord)：

Producer确保在消息被序列化以及计算分区前调用该方法

该方法封装进KafkaProducer.send方法中，即它运行在用户主线程中

用户可以在该方法中对消息做任何操作，但最好保证不要修改消息所属的topic和分区，否则会影响目标分区的计算

（3）onAcknowledgement(RecordMetadata, Exception)：

该方法会在消息被应答或消息发送失败时调用，并且通常都是在producer回调逻辑触发之前onAcknowledgement运行在producer的IO线程中，因此不要在该方法中放入很重的逻辑，否则会拖慢producer的消息发送效率

（4）close：

关闭interceptor，主要用于执行一些资源清理工作

Interceptor可能被运行在多个线程中，因此在具体实现时用户需要自行确保线程安全

倘若指定了多个interceptor，则producer将按照指定顺序调用它们，并仅仅是捕获每个interceptor可能抛出的异常记录到错误日志中而非在向上传递

### 八、kafka Streams (轻量级，功能性弱) ###

#### 1 Kafka Streams概括 ####

Kafka Streams。Apache Kafka开源项目的一个组成部分。是一个功能强大，易于使用的库。用于在Kafka上构建高可分布式、拓展性，容错的应用程序。

#### 2 Kafka Streams特点 ####

1）功能强大 高扩展性，弹性，容错

2）轻量级 无需专门的集群 一个库，而不是框架

3）完全集成 易于集成到现有的应用程序

4）实时性 毫秒级延迟 并非微批处理 窗口允许乱序数据 允许迟到数据

#### 3 为什么要有Kafka Stream ####

当前已经有非常多的流式处理系统，最知名且应用最多的开源流式处理系统有Spark Streaming和Apache Storm。Apache Storm发展多年，应用广泛，提供记录级别的处理能力，当前也支持SQL on Stream。而Spark Streaming基于Apache Spark，可以非常方便与图计算，SQL处理等集成，功能强大，对于熟悉其它Spark应用开发的用户而言使用门槛低。另外，目前主流的Hadoop发行版，如Cloudera和Hortonworks，都集成了Apache Storm和Apache Spark，使得部署更容易。

既然Apache Spark与Apache Storm拥用如此多的优势，那为何还需要Kafka Stream呢？

主要有如下原因：

第一，Spark和Storm都是流式处理框架，而Kafka Stream提供的是一个基于Kafka的流式处理类库。框架要求开发者按照特定的方式去开发逻辑部分，供框架调用。开发者很难了解框架的具体运行方式，从而使得调试成本高，并且使用受限。而Kafka Stream作为流式处理类库，直接提供具体的类给开发者调用，整个应用的运行方式主要由开发者控制，方便使用和调试

第二，虽然Cloudera与Hortonworks方便了Storm和Spark的部署，但是这些框架的部署仍然相对复杂。而Kafka Stream作为类库，可以非常方便的嵌入应用程序中，它对应用的打包和部署基本没有任何要求

第三，就流式处理系统而言，基本都支持Kafka作为数据源。例如Storm具有专门的kafka-spout，而Spark也提供专门的spark-streaming-kafka模块。事实上，Kafka基本上是主流的流式处理系统的标准数据源。换言之，大部分流式系统中都已部署了Kafka，此时使用Kafka Stream的成本非常低

第四，使用Storm或Spark Streaming时，需要为框架本身的进程预留资源，如Storm的supervisor和Spark on YARN的node manager。即使对于应用实例而言，框架本身也会占用部分资源，如Spark Streaming需要为shuffle和storage预留内存。但是Kafka作为类库不占用系统资源

第五，由于Kafka本身提供数据持久化，因此Kafka Stream提供滚动部署和滚动升级以及重新计算的能力

第六，由于Kafka Consumer Rebalance机制，Kafka Stream可以在线动态调整并行度

### 九、kafka & flume ###

#### 1 比较 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwOTI5OTIx_size_16_color_FFFFFF_t_70 4]

flume：cloudera公司研发

适合多个生产者 收集多台服务器产生的数据，一个生产者一个Agent

适合下游数据消费者不多的情况 多一个消费者多一个Channel，每个Channel数据一致，浪费内存

适合数据安全性要求不高的操作 基于内存

适合与Hadoop生态圈对接的操作

kafka：linkedin公司研发

适合数据下游消费众多的情况 副本和消费者没有关系，增加消费者就阔以

适合数据安全性要求较高的操作，支持replication 存储在磁盘

因此我们常用的一种模型是：

线上数据 --> flume --> kafka --> flume(根据情景增删该流程) --> HDFS

因为多个线上数据进行汇总后放入一个flume中，之后的业务可能会有很多的下游消费者，因此flume不适合直连多个消费者，因此需要将数据导入到kafka中，由kafka进行多个消费者消费数据的情况

#### 2 集成配置(暂无) ####

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwOTI5OTIx_size_16_color_FFFFFF_t_70]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/d64facdaf9894579baa48423d5834e81.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwOTI5OTIx_size_16_color_FFFFFF_t_70 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/769a7bb7d3354825b0e008b1359e772c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwOTI5OTIx_size_16_color_FFFFFF_t_70 2]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/5d5c3a1d50f34782b593e16976a3e134.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwOTI5OTIx_size_16_color_FFFFFF_t_70 3]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/6108d8b205e145e992c82876cf048187.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwOTI5OTIx_size_16_color_FFFFFF_t_70 4]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/d5b49f33beaa4393a2e67d42ac175244.png