kafka整理

左手的ㄟ右手 2022-11-08 14:15 101阅读 0赞

1.1 消费定义

*  同步处理  
    直连，同步在线，A和B系统之间通信，同步在线状态，等待B系统反馈。直接连接；相当于一个线程走下来； 耦合较高
 *  异步处理  
    填写注册信息，创建订单信息，入库仓管信息，状态变更信息，推送短信信息。这些状态变更通过消息进行发送，由中间件管理，在Mq缓存，中间件隔离，如果A系统挂了，可以等待A系统启动重新消费。指针指向；

1.2

主要功能： 异步解耦， 高并发导致的服务器压力，不能同步处理，可以进行流量削峰，不需要过度加机器；如果并发量较大，加减机器不会影响项目之间机器切换导致的资源过剩。

1.3

消息队列的两种模式：一对一，消费者主动拉去消息；消息收到之后消息消息清楚；一对多，消息不回清除，消息发到topc上，所有订阅此消息的消费者，都会被推送到消息；

说到订阅发布，tipic存储的消息，会被存起来；

订阅发布模式，又分为两种，一种是主动推送，一种是主动拉取，基于拉取方面，解决了消费者跟生产速宽不一致问题

举例，微信公众号，只有订阅了，就会必推给你。所以可能会导致有些消息不是你想要的，所以可以定义消息过滤和tags进行分组订阅，但是如果生产过快，消费过慢，队列堆积过大，可能会出现异常处理问题，如果为了保持带宽，又会造成服务器压力；

但是主动拉去，也是有缺点，需要不停的问生产者，有没有消息，涉及到一个轮询资源浪费问题，所以性能跟速度，并不好解决；因为需要维护长轮询。

kafka是维护一个消息推送过来到一个消费组，进行消费分配策略，然后由消费者去主动拉去数据

1.4

**kafka 主要成员**

生产者，消费者，kafka集群，zk集群，消费组，broker kafka集群中一个kafka的进程

**整体架构**  
生产 - cluster - 消费  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDU1MDQ5MA_size_16_color_FFFFFF_t_70]

breaker 内部结构  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDU1MDQ5MA_size_16_color_FFFFFF_t_70 1]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDU1MDQ5MA_size_16_color_FFFFFF_t_70 2]  
生产者发送消息到服务器容错  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDU1MDQ5MA_size_16_color_FFFFFF_t_70 3]

服务器kafka 与消费者交互  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDU1MDQ5MA_size_16_color_FFFFFF_t_70 4]  
po 到p4 属于一个topic 的多个partition  
一个消费者可以消费多个分区，但是一个分区只能给一个消费者去消费  
副本同步  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDU1MDQ5MA_size_16_color_FFFFFF_t_70 5]

1.5

思考一个问题：多个消息发送到kafka集群里面，如果一个服务器上的topic服务器挂了怎么处理：

为了处理这个情况，所以多个一个topic会在集群里面共享在多个机器上面存储，这样才能保证其中一台机器挂了，还能保证这个topic正常发送；

再思考一个问题，如果把topic 放在多个机器，如果一个消息过来，存在那个机器上面，如果存储消息的服务器挂了怎么处理；

partition，对于消息进行分区，这样可以保证机器性能，一条消息发送机器A broker，partition分区，一条在机器B broker partition 分区。这样可以进行负载均衡，保证机器性能，同时一个消息为了防止挂机丢失，都有备份，并且在机器中选取一个leader，leader进行消息主要的存储发送，并且定义follower 进行备份处理。保证机器挂掉，备份消息存在，消息不丢失问题；

思考一个问题：如果想维护一个kafka集群，如何搭建，如何能让他们集群在一起，如果消费挂了，消费了mq上面的消息第四条，重新启动消费下一条，这个记录需要怎么处理；

使用zk 注册消息把这些机器，放在一起，相互注册，搭建成一个集群。关于消费到那一条，由zk存储一下指针，帮助kafka知道，消费者消费到指针的第几条；这个kafka 后面高版本放在kafka维护，后面会聊一下为什么要放到kafka，举个例子，每次挂掉重新找，或者消费第几条，每次都要去zk去找记录，项目一般不喜欢耦合，所以后面会放到kafka，放到主系统。

完成以上，整个消息中间件大致核心功能完善；

1.6

安装部署

进入官网下一下kafka，kafka也交给了apache管理，所以官网自然是kafka.apache.org

直接把包下下来， 包就算.tgz结尾的压缩包

解压一下压缩包 -c是解压的路径

tar -zxvf kafka_2.11-0.11.0.0.taz -C /opt/xu/

可以看一下

cd ../xu/

可以通过mv 改一下名字 mv kafka … kafka

cd kafka

可以看到配置文件config

cd config

可以看到服务配置，消费配置，生产配置，日志配置，zk配置

服务配置一下

broker.id  配置一下kafka这台机器的唯一id 默认是0
    
    delete.topic.endble  默认不能删除主题 false
    num.network.threads 处理网络线程数量
    log.dirs   服务日志存放文件
    num.partition 当前机器分区数量
    zookeeper.connect  zk集群地址，多台,号分割

如果配置了堕胎zk集群地址的机器可以进行分发一下堕胎服务器的服务配置

xsync kafak/

启动一下  
命令都在bin文件里面

kafka-servier-start.sh  
topic 管理在 kafka-topic.sh

启动服务

bin  kafka-servier-start.sh config /servier.properties

1.7  
聊一下kafka工作流程

生产者，消费者，kafka集群，多个broker。当没有topic的时候会根据服务Service配置文件进行创建topic ，每一个分区，从头到尾的偏移量维护，举例A2条B2条C2条，那么偏移量三台机器，每台偏移两条，每个partition只能保证区内有序。  
同样副本是不会在同一台机器，会在不同机器，错开进行消费记录备份。  
消息会存储到kafka log存储，sgement，最大一个G超过一个G就会再生成一个log， 生成一个sgement片段。如果生产过多，这些数据又该如何去找，会定义一个index 进行索引。保证定位数据问题

1.8 聊一聊AR、ISR、OSR

*  AR 分区中所有副本统称为 AR；
 *  ISR 所有与leader副本保持一定程度的同步的副本（包括leader）组成  
    ISR是AR的一个子集； replica.lag.time.max.ms（默认10秒）：follower副本能落后leader 副本的最长时间间隔；如果超时就会被移除
 *  OSR 与leader副本同步滞后太多的副本（不包括leader）组成OSR；（如果后面追上leader进度的话，会被重新加回ISR）

kafka的ISR机制被成为“不丢消息”机制。在说ISR机制前，先聊一下kafka的副本。  
kafka的Replica

*  1.kafka的topic可以设置有N个副本（replica），副本数最好要小于broker的数量，也就是要保证一个broker上的replica最多有一个，所以可以用broker id指定Partition replica。
 *  2.创建副本的单位是topic的分区，每个分区有1个leader和0到多个follower，我们把多个replica分为Lerder replica和follower replica。
 *  3.当producer在向partition中写数据时，根据ack机制，默认ack=1，只会向leader中写入数据，然后leader中的数据会复制到其他的replica中，follower会周期性的从leader中pull数据，但是对于数据的读写操作都在leader replica中，follower副本只是当leader副本挂了后才重新选取leader，follower并不向外提供服务。  
    kafka的“同步”  
    kafka不是完全同步，也不是完全异步，是一种特殊的ISR（In Sync Replica）  
    1.leader会维持一个与其保持同步的replica集合，该集合就是ISR，每一个partition都有一个ISR，它时有leader动态维护。  
    2.我们要保证kafka不丢失message，就要保证ISR这组集合存活（至少有一个存活），并且消息commit成功。

所以我们判定存活的概念时什么呢？分布式消息系统对一个节点是否存活有这样两个条件判断：第一个，节点必须维护和zookeeper的连接，zookeeper通过心跳机制检查每个节点的连接；第二个，如果节点时follower，它必要能及时同步与leader的写操作，不是延时太久。  
如果满足上面2个条件，就可以说节点时“in-sync“（同步中的）。leader会追踪”同步中的“节点，如果有节点挂了，卡了，或延时太久，那么leader会它移除，延时的时间由参数replica.log.max.messages决定，判断是不是卡住了，由参数replica.log.time.max.ms决定。  
kafka的commit是由ack机制决定的

*  \-1 为同步所有副本之后再跟生产者，说ok我干完了
 *  1 老大说好了，就会跟生产说，我可以了
 *  0 就是生产者说我发出去了，你拿没拿到我不管了  
    不过1和-1 可能会出现消息重复，比如生产者发一条消息到kafka ，老大接受完，写完了，准备发ack我ok 挂了，副本就会成为一个老大，生产者说我没发吗又发了一条。消息重复

1.9 聊一下 leader挂了 一致性问题 HW LEO

举个例子，比如leader挂了，子备份成为leader，接受消息后，leader活了。  
举例

字面意思，HW 消费者可见的最大offset，LEO 最小offset每个副本最大的offset

leader活了，之后，自己只有10条数据，副本有15条。出现数据不一致问题，同步到15条还是10条，进行设置。hw 就算如果15是老大了， 消费可见是10条，截取到10条，leo是多退少补，截取到15条。

简单说一下HW 和 LEO 定义  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDU1MDQ5MA_size_16_color_FFFFFF_t_70 6]

*  HW是 High Watermark的缩写，也叫高水位，标识了一个特定的消息偏移量（offset） hw高水位是指在一个分区中 ，可以被消费者可见的消息的最大
 *  LEO是 Log End Offset 的缩写，标识日志末端位移，也标识副本写入下一条消息的位移；本分去最大的数据高位

两个方面hw保证了消费的一致性，leo保证了存储的完整性

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDU1MDQ5MA_size_16_color_FFFFFF_t_70]: /images/20221023/1b3a80fe57bd4f69bba22cd05b6c29d7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDU1MDQ5MA_size_16_color_FFFFFF_t_70 1]: /images/20221023/90ee72b578564a639acd5e5bb443ff02.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDU1MDQ5MA_size_16_color_FFFFFF_t_70 2]: /images/20221023/19e003ea577548dc9eec32ffbb42d702.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDU1MDQ5MA_size_16_color_FFFFFF_t_70 3]: /images/20221023/55bcb39db68f4e13b5eb4e131d14acb6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDU1MDQ5MA_size_16_color_FFFFFF_t_70 4]: /images/20221023/39af2d1a9cef44fe952f126a5704c45a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDU1MDQ5MA_size_16_color_FFFFFF_t_70 5]: /images/20221023/8851d8a299ed49b2a72d3a8628089bfc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NDU1MDQ5MA_size_16_color_FFFFFF_t_70 6]: /images/20221023/46fde595c5d84f9abe801d3cdabc6a00.png