Kafka安装及架构

kafka的特点

• 高吞吐量、低延迟：kafka每秒可以处理几十万条消息，它的延迟最低只有几毫秒，每个topic可以分多个partition, 由多个consumer group 对partition进行consume操作。
• 可扩展性：kafka集群支持热扩展
• 持久性、可靠性：消息被持久化到本地磁盘，并且支持数据备份防止数据丢失
• 容错性：允许集群中有节点失败（若副本数量为n,则允许n-1个节点失败）
• 高并发：支持数千个客户端同时读写

安装

1. 上传安装包
2. 解压

 tar -zxvf kafka_2.11-2.2.2.tgz tar-C /opt/apps/

3.修改配置文件

（1）进入配置文件目录[root@linux01 apps]# cd kafka_2.12-2.3.1/config（2）编辑配置文件vi server.properties#为依次增长的：0、1、2、3、4，集群中唯一 idbroker.id=0 #数据存储的⽬录 log.dirs=/opt/data/kafkadata#底层存储的数据（日志）留存时长（默认7天）log.retention.hours=168#底层存储的数据（日志）留存量（默认1G） log.retention.bytes=1073741824#指定zk集群地址zookeeper.connect=linux01:2181,linux02:2181,linux03:2181

4. 分发安装包

foriin {2..3} do scp-rkafka_2.11-2.2.2linux0$i:$PWD done

5.环境变量配置

vi /etc/profileexport KAFKA_HOME=/opt/apps/kafka_2.11-2.2.2export PATH=$PATH:$KAFKA_HOME/bin source /etc/profile# 注意：还需要分发环境变量

6. 一键启停脚本：

#!/bin/bashcase $1 in"start"){for i in linux01 linux02 linux03doecho ---------- kafka $i 启动 ------------ssh $i "source /etc/profile;/opt/app/kafka2.4.1/bin/kafka-server-start.sh -daemon /opt/app/kafka2.4.1/config/server.properties"done};;"stop"){for i in linux01 linux02 linux03doecho ---------- kafka $i 停止 ------------ssh $i "source /etc/profile;/opt/app/kafka2.4.1/bin/kafka-server-stop.sh "done};;esac

应用场景

• 日志收集：一个公司可以用kafka可以收集各种服务的log，通过kafka以统一接口服务的方式开放给各种consumer，例如hadoop、HBase、Solr等。
• 消息系统：解耦和生产者和消费者、缓存消息等。
• 用户行为跟踪：kafka经常被用来记录web用户或者app用户的各种活动，如浏览网页、搜索、点击等活动，这些活动信息被各个服务器发布到kafka的topic中，然后订阅者通过订阅这些topic来做实时的监控分析，或者装载到hadoop、数据仓库中做离线分析和挖掘。
• 运营指标：kafka也经常用来记录运营监控数据。包括收集各种分布式应用的数据，生产各种操作的集中反馈，比如报警和报告。
• 作为流式处理的数据源：比如spark streaming和 Flink

kafka架构

主题topic和分区partition

• topic
  Kafka中存储数据的逻辑分类；你可以理解为数据库中“表”的概念；
  比如，将app端日志、微信小程序端日志、业务库订单表数据分别放入不同的topic
• partition分区（提升kafka吞吐量）
  topic中数据的具体管理单元；（你可以理解为hbase中表的“region”概念）
• 每个partition由一个kafka broker服务器管理；
• 每个topic 可以划分为多个partition，分布到多个broker上管理；
• 每个partition都可以有多个副本；保证数据安全

分区对于 kafka 集群的好处是：实现topic数据的负载均衡。提高写入、读出的并发度，提高吞吐量。

• 分区副本replica
  每个topic的每个partition都可以配置多个副本（replica），以提高数据的可靠性；
  每个partition的所有副本中，必有一个leader副本，其他的就是follower副本（observer副本）；follower定期找leader同步最新的数据；对外提供服务只有leader；
• 分区follower
  partition replica中的一个角色，它通过心跳通信不断从leader中拉取、复制数据（只负责备份）。
  如果leader所在节点宕机，follower中会选举出新的leader；
• 消息偏移量offset
  partition内部每条消息都会被分配一个递增id（offset）；通过offset可以快速定位到消息的存储位置；
  kafka 只保证按一个partition中的消息的顺序，不保证一个 topic的整体（多个partition 间）的顺序。
  我们在说到偏移量的时候，是哪一个topic的哪一个分区的哪一个，偏移量他的数据只能追加，不能被修改
  
  自我推导设计：
• kafka是用来存数据的；
• 现实世界数据有分类，所以存储系统也应有数据分类管理功能，如mysql的表；kafka有topic；
• 如一个topic的数据全部交给一台server存储和管理，则读写吞吐量有限；
• 所以，一个topic的数据应该可以分成多个部分（partition）分别交给多台server存储和管理；
• 如一台server宕机，这台server负责的partition将不可用，所以，一个partition应有多个副本；
• 一个partition有多个副本，则副本间的数据一致性难以保证，因此要有一个leader统领读写；
• 一个leader万一挂掉，则该partition又不可用，因此还要有leader的动态选举机制；
• 集群有哪些topic，topic有哪几个分区，server在线情况，等等元信息和状态信息需要在集群内部及客户端之间共享，则引入了zookeeper；
• 客户端在读取数据时，往往需要知道自己所读取到的位置，因而要引入消息偏移量维护机制；
  broker服务器：一台 kafka服务器就是一个broker。一个kafka集群由多个 broker 组成。
  生产者producer：消息生产者，就是向kafka broker发消息的客户端。
  消费者consumer
• consumer ：消费者，从kafka broker 取消息的客户端。
• consumer group：消费组，单个或多个consumer可以组成一个消费组；
  消费组是用来实现消息的广播（发给所有的 consumer）和单播（发给任意一个 consumer）的手段；
  

kafka的数据存储结构

物理存储目录结构 __consumer_offset
存储目录 名称规范： topic名称-分区号

• 数据文件 名称规范：
  生产者生产的消息会不断追加到log文件末尾，为防止log文件过大导致数据定位效率低下，Kafka采取了分片和索引机制

1. 每个partition的数据将分为多个segment存储
2. 每个segment对应两个文件：“.index”文件和“.log”文件。
   
   index和log文件以当前segment的第一条消息的offset命名。

index索引文件中的数据为： 消息offset -> log文件中该消息的物理偏移量位置；
Kafka 中的索引文件以稀疏索引（ sparse index ）的方式构造消息的索引，它并不保证每个消息在索引文件中都有对应的索引；每当写入一定量（由 broker 端参数 log.index.interval.bytes 指定，默认值为 4096 ，即 4KB ）的消息时，偏移量索引文件和时间戳索引文件分别增加一个偏移量索引项和时间戳索引项，增大或减小 log.index.interval.bytes的值，对应地可以缩小或增加索引项的密度；
查询指定偏移量时，使用二分查找法来快速定位偏移量的位置。

消息message存储结构

在客户端编程代码中，消息的封装类有两种：ProducerRecord、ConsumerRecord；
简单来说，kafka中的每个massage由一对key-value构成；
Kafka中的message格式经历了3个版本的变化了：v0 、 v1 、 v2

各个字段的含义介绍如下：

• crc：占用4个字节，主要用于校验消息的内容；
• magic：这个占用1个字节，主要用于标识日志格式版本号，此版本的magic值为1
• attributes：占用1个字节，这里面存储了消息压缩使用的编码以及Timestamp类型。目前Kafka 支持 gzip、snappy 以及 lz4（0.8.2引入） 三种压缩格式；[0,1,2]三位bit表示压缩类型。[3]位表示时间戳类型（0，create time；1，append time），[4,5,6,7]位保留；
• key length：占用4个字节。主要标识 Key的内容的长度；
• key：占用 N个字节，存储的是 key 的具体内容；
• value length：占用4个字节。主要标识 value 的内容的长度；
• value：value即是消息的真实内容，在 Kafka 中这个也叫做payload。

日志分段切分条件

日志分段文件切分包含以下4个条件，满足其一即可：

1. 当前日志分段文件的大小超过了broker端参数 log.segment.bytes 配置的值。
   log.segment.bytes参数的默认值为 1073741824，即1GB
2. 当前日志分段中消息的最小时间戳与当前系统的时间戳的差值大于log.roll.ms或log.roll.hours参数配置的值。如果同时配置了log.roll.ms和log.roll.hours 参数，那么 log.roll.ms 的优先级高，默认情况下，只配置了log.roll.hours参数，其值为168,即7天。
3. 偏移量索引文件或时间戳索引文件的大小达到 broker 端参数 log.index.size.max.bytes 配置的值。
   log.index.size .max.bytes的默认值为 10485760，即10MB
4. 追加的消息的偏移量与当前日志分段的起始偏移量之间的差值大于Integer.MAX_VALUE, 即要追加的消息的偏移量不能转变为相对偏移量（offset – baseOffset > Integer.MAX_VALUE）。

什么是Controller

Controller作为Kafka集群中的核心组件，它的主要作用是在 Apache ZooKeeper 的帮助下管理和协调整个 Kafka 集群。
Controller与Zookeeper进行交互，获取与更新集群中的元数据信息。其他broker并不直接与zookeeper进行通信，而是与 Controller 进行通信并同步Controller中的元数据信息。
Kafka集群中每个节点都可以充当Controller节点，但集群中同时只能有一个Controller节点。
在Kafka集群中会有一个或者多个broker，其中有一个broker会被选举为控制器（Kafka Controller），它负责维护整个集群中所有分区和副本的状态及分区leader的选举。
当某个分区的leader副本出现故障时，由控制器负责为该分区选举新的leader副本。当检测到某个分区的ISR集合发生变化时，由控制器负责通知所有broker更新其元数据信息。当使用kafka-topics.sh脚本为某个topic增加分区数量时，同样还是由控制器负责分区的重新分配。
Kafka中的控制器选举的工作依赖于Zookeeper，成功竞选为控制器的broker会在Zookeeper中创建/controller这个临时（EPHEMERAL）节点，此临时节点的内容参考如下：
{“version”:1,“brokerid”:0,“timestamp”:“1529210278988”}
其中version在目前版本中固定为1，brokerid表示成为控制器的broker的id编号，timestamp表示竞选成为控制器时的时间戳。
在任意时刻，集群中有且仅有一个控制器。每个broker启动的时候会去尝试去读取zookeeper上的/controller节点的brokerid的值，如果读取到brokerid的值不为-1，则表示已经有其它broker节点成功竞选为控制器，所以当前broker就会放弃竞选；如果Zookeeper中不存在/controller这个节点，或者这个节点中的数据异常，那么就会尝试去创建/controller这个节点，当前broker去创建节点的时候，也有可能其他broker同时去尝试创建这个节点，只有创建成功的那个broker才会成为控制器，而创建失败的broker则表示竞选失败。每个broker都会在内存中保存当前控制器的brokerid值，这个值可以标识为activeControllerId。

controller的职责

• 监听partition相关变化

对Zookeeper中的/admin/reassign_partitions节点注册PartitionReassignmentListener，用来处理分区重分配的动作。对Zookeeper中的/isr_change_notification节点注册IsrChangeNotificetionListener，用来处理ISR集合变更的动作。对Zookeeper中的/admin/preferred-replica-election节点添加PreferredReplicaElectionListener，用来处理优先副本选举。

• 监听topic增减变化

对Zookeeper中的/brokers/topics节点添加TopicChangeListener，用来处理topic增减的变化；对Zookeeper中的/admin/delete_topics节点添加TopicDeletionListener，用来处理删除topic的动作。

• 监听broker相关的变化

对Zookeeper中的/brokers/ids/节点添加BrokerChangeListener，用来处理broker增减的变化。

• 更新集群的元数据信息

从Zookeeper中读取获取当前所有与topic、partition以及broker有关的信息并进行相应的管理。对各topic所对应的Zookeeper中的/brokers/topics/[topic]节点添加PartitionModificationsListener，用来监听topic中的分区分配变化。并将最新信息同步给其他所有broker。

• 启动并管理分区状态机和副本状态机。
• 如果参数auto.leader.rebalance.enable设置为true，则还会开启一个名为“auto-leader-rebalance-task”的定时任务来负责维护分区的leader副本的均衡。

分区的负载分布

客户端请求创建一个topic时，每一个分区副本在broker上的分配，是由集群controller来决定；
结论：里面会创建出来两个随机数
第一个随机数确定0号分区leader的位置，往后1号分区2号分区的leader依次往后顺延1
第二个随机数确定每个分区的第一个副本的位置 在leader所在机器上往后顺延（随机数+1）台机器，
该台机器就是第一个副本的位置，剩余副本依次往后顺延1

举例：broker_id = 0~19 一共20台机器分区数20,副本数10第一个随机数：19第二个随机数：0(0,ArrayBuffer(19, 0, 1, 2, 3, 4, 5, 6, 7, 8))(1,ArrayBuffer(0, 1, 2, 3, 4, 5, 6, 7, 8, 9))(2,ArrayBuffer(1, 2, 3, 4, 5, 6, 7, 8, 9, 10))(3,ArrayBuffer(2, 3, 4, 5, 6, 7, 8, 9, 10, 11))(4,ArrayBuffer(3, 4, 5, 6, 7, 8, 9, 10, 11, 12))(5,ArrayBuffer(4, 5, 6, 7, 8, 9, 10, 11, 12, 13))(6,ArrayBuffer(5, 6, 7, 8, 9, 10, 11, 12, 13, 14))(7,ArrayBuffer(6, 7, 8, 9, 10, 11, 12, 13, 14, 15))(8,ArrayBuffer(7, 8, 9, 10, 11, 12, 13, 14, 15, 16))(9,ArrayBuffer(8, 9, 10, 11, 12, 13, 14, 15, 16, 17))(10,ArrayBuffer(9, 10, 11, 12, 13, 14, 15, 16, 17, 18))(11,ArrayBuffer(10, 11, 12, 13, 14, 15, 16, 17, 18, 19))(12,ArrayBuffer(11, 12, 13, 14, 15, 16, 17, 18, 19, 0))(13,ArrayBuffer(12, 13, 14, 15, 16, 17, 18, 19, 0, 1))(14,ArrayBuffer(13, 14, 15, 16, 17, 18, 19, 0, 1, 2))(15,ArrayBuffer(14, 15, 16, 17, 18, 19, 0, 1, 2, 3))(16,ArrayBuffer(15, 16, 17, 18, 19, 0, 1, 2, 3, 4))(17,ArrayBuffer(16, 17, 18, 19, 0, 1, 2, 3, 4, 5))(18,ArrayBuffer(17, 18, 19, 0, 1, 2, 3, 4, 5, 6))(19,ArrayBuffer(18, 19, 0, 1, 2, 3, 4, 5, 6, 7))

其分布策略源码如下：

private def assignReplicasToBrokersRackUnaware(nPartitions: Int, //分区的个数 10replicationFactor: Int,//副本的个数5 brokerList: Seq[Int],//broker的集合8 0~7fixedStartIndex: Int//默认值是-1固定开始的索引位置startPartitionId: Int): Map[Int, Seq[Int]] //默认值是-1 分区开始的位置= {val ret = mutable.Map[Int, Seq[Int]]()val brokerArray = brokerList.toArrayval startIndex = if (fixedStartIndex >= 0) {fixedStartIndex}else {rand.nextInt(brokerArray.length)}var currentPartitionId = math.max(0, startPartitionId) var nextReplicaShift = if (fixedStartIndex >= 0) {fixedStartIndex }else {rand.nextInt(brokerArray.length)}for (_ <- 0 until nPartitions) {if (currentPartitionId > 0 && (currentPartitionId % brokerArray.length == 0)){nextReplicaShift += 1}val firstReplicaIndex = (currentPartitionId + startIndex) % brokerArray.length val replicaBuffer = mutable.ArrayBuffer(brokerArray(firstReplicaIndex))for (j <- 0 until replicationFactor - 1) {replicaBuffer += brokerArray(replicaIndex(firstReplicaIndex, nextReplicaShift, j, brokerArray.length))}ret.put(currentPartitionId, replicaBuffer)currentPartitionId += 1}ret} private def replicaIndex(firstReplicaIndex: Int, secondReplicaShift: Int, replicaIndex: Int, nBrokers: Int): Int = {val shift = 1 + (secondReplicaShift + replicaIndex) % (nBrokers - 1)(firstReplicaIndex + shift) % nBrokers}

• 副本因子不能大于 Broker 的个数；
  报错：Replication factor: 4 larger than available brokers：3
• partition_0的第1个副本（leader副本）放置位置是随机从 brokerList 选择的；
• 其他分区的第1个副本（leader）放置位置相对于paritition_0分区依次往后移（也就是如果我们有5个 Broker，5个分区，假设partition0分区放在broker4上，那么partition1将会放在broker5上；patition2将会放在broker1上；partition3在broker2，依次类）；
• 各分区剩余的副本相对于分区前一个副本偏移随机数nextReplicaShift+1，然后后面的副本依次加1

分区Leader的选举机制

分区 leader 副本的选举由控制器controller负责具体实施。
当创建分区（创建主题或增加分区都有创建分区的动作）或Leader下线（此时分区需要选举一个新的leader上线来对外提供服务）的时候都需要执行 leader 的选举动作。
选举策略：按照 ISR集合中副本的顺序查找第一个存活的副本，并且这个副本在 ISR 集合中