Apache Kafka 源码剖析

时间:2023-03-10 07:11:04
Apache Kafka 源码剖析

Getting Start

下载

优点和应用场景

  • Kafka消息驱动,符合发布-订阅模式,优点和应用范围都共通
  • 发布-订阅模式优点
    1. 解耦合 : 两个应用不需要相互调用
    2. 可扩展性 : 消费者的个数可实时扩展
    3. 实时性 : 消费者能实时的获取生产者发布的事件
    4. 高效 :减少由于多个消费者请求数据造成的数据计算带来的资源消耗
    5. 异步通讯 :发布-订阅模式是天生的异步通讯
  • Kafka其他优点
    1. 持久化 : 消息丢失的可控性极高
    2. 高性能
    3. 顺序性
  • 发布-订阅模式应用范围
    1. 适合数据一被生产,就需要被处理的情况
    2. 适合数据具有潜在消费者的情况
    3. 适合无论有没有消费者,数据都在生产的情况
    4. 不适合对数据的处理时间有特殊限定的情况
  • 应用场景
    1. 最为消息中间件,实现消息队列和消息的发布-订阅,消息驱动的服务
    2. 数据总线,一对多的模式
    3. 日志收集,消息中间件的一种应用
    4. 数据库主从同步

核心概念

  • Broker
    1. 一个Kafka server就是一个Broker
    2. 一般情况下,一个Broker独占一台服务器,发挥微服务的优势
    3. 服务器资源有限的情况下,需要设计出Broker/Topic/Partition/Replica的最优分配策略,从而充分利用服务器资源
    4. 一个broker可以有多个topic
  • Topic
    1. 存储消息的逻辑上的消息集合
    2. 每个Topic有多个分区
  • 分区 Partition
    1. 同一个Topic的不同分区分配在不同Broker上,也就是一个分区一个服务器
    2. 不同Topic的分区可以共享一个服务器
    3. 同一个分区的消息是有序的,通过维护offset实现
    4. 相同key的消息会被发布到同一个分区
    5. 同一个分区的消息会被一个消费组里固定的一个消费者独占消费
    6. 通过增加分区来增加并行处理能力
    7. 每个分区可以有多个副本
  • 消费组 Consumer Group
    1. 实现一个消息只被同组的一个消费者独占消费
    2. 消费组里的消费者有变化的时候会触发Rebalance操作重新分配分区与消费者的对应关系
    3. Rebalance操作实现了分区消费的故障转移
    4. 通过增加分区和消费组里的消费者数量来水平扩展,理想情况一对一,也可以一对多,最好不要多对一,造成浪费
  • 副本 Replica
    1. 同一个分区的不同副本分配在不同Broker上,但是这些Broker可以是在同一台服务器上,也可以不是
    2. 副本是一个热备份设计,会选举一个作为Leader,提供对外服务
    3. Fllower副本批量的从Leader副本同步消息
  • HW & LEO
    1. HW是所有ISR副本都有的最新offset,HW之前的消息在所有副本中都存在,HW由Leader副本维护
    2. 所有消费者只能获取HW之前的消息,这样保证了Leader副本不可用的情况下,所有消费者的状态是一致的
    3. LEO是每个副本各自的最新offset
  • ISR集合
    1. 满足两个条件的副本会被选入ISR可用副本集合
      1. 副本与Zookeeper连接
      2. 副本的LEO与Leader副本的LEO差值不超过阈值
    2. ISR集合保证了Kafka不会被故障副本拖累,也保证了Leader的HW与LEO的差值在阈值内
  • 生产者
    1. 异步提交
      1. acks=0 : 生产者只管提交,不会等待Leader副本返回,不保证数据不丢失
    2. 同步提交
      1. acks=1 : 默认设置,生产者等待Leader副本返回成功,保证数据在Leader中部丢失,但是不保证重新选举后数据不丢失
    3. 同步复制
      1. acks=all : 生产者等待所有副本同步消息后才算提交成功,保证数据不丢失,性能低
  • Log
    1. 一个副本对应一个Log,用于持久化数据,Kafka采用顺序读写的方式,性能高
    2. 一个Log里有多个Segment,每个Segment有一个日志文件和一个索引文件
    3. 日志文件的大小有限制,超出后会生成新的Segment
    4. 日志消息保留策略有两种
      1. 消息的保留时间超过指定时间,可以被删除
      2. Topic的存储满,可以开始删除最旧的消息
      3. 保留策略可以全局配置,也可以按Topic配置
    5. 日志压缩
      1. 开启日志压缩后,相同的key会被定期合并,只保留最新的value

Kafka/zookeeper 命令

  1. 启动Zookeeper
    1. ./zookeeper-server-start.sh ../config/zookeeper.properties
  2. 启动Kafka
    1. ./kafka-server-start.sh ../config/server.properties
  3. 查看Topic
    1. ./kafka-topics.sh --list --zookeeper localhost:9860
  4. 删除Topic
    1. ./kafka-topics.sh --delete --zookeeper localhost:9860 /kafka --topic test
    2. 不会立马删除topic
  5. 查看Topic的详细信息
    1. ./kafka-topics.sh --zookeeper localhost:9860 --topic test--describe
  6. 查看zk信息
    1. ./zookeeper-shell.sh 127.0.0.1:9860
  7. 生产数据
    1. ./kafka-console-producer.sh --broker-list cvatap3d.nam.nsroot.net:9801 --topic midcurve-ds
  8. 消费数据
    1. ./kafka-console-consumer.sh --zookeeper localhost:9860 --topic midcurve-ds-subscribe --from-beginning

Kafka集群

zookeeper集群配置 : zookeeper.properties

  • clientPort=2180
    1. 端口号
  • dataDir=/tmp/zookeeper
    1. 集群信息记录目录,清空目录可以重置zookeeper
    2. 如果需要在同一台server上启动多个node,这个路径必须不同
  • tickTime=2000
    1. zookeeper副本与leader之间维护心跳的频率
  • initLimit=5
    1. zookeeper的leader初始化连接follower时等待多少个tickTime时间的心跳,超时副本连接失败
  • syncLimit=2
    1. leader与follower之间发送消息,请求和应答超时是多少个tickTime
  • server.0=cvatap3d.nam.nsroot.net:2888:3888
  • server.1=cvatap3d.nam.nsroot.net:2889:3889
  • server.2=cvatap3d.nam.nsroot.net:2890:3890
    1. 第一个启动的为leader
    2. zookeeper集群数量必须是基数3,5,7...
    3. 0,1,2是服务id,需要在对应的dataDir=/tmp/zookeeper下面创建myid文件,内容就是服务id,比如0
    4. ip或者host都可以
    5. 后面两个端口是zookeeper内部通讯使用
      1. 第一个端口是用于副本与Leader建立TCP连接
      2. 第二个端口是用于Leader选举的TCP端口

Kafka配置 : server.properties

  • broker.id=0
    1. 同一个zookeeper集群下的broker的id必须唯一
  • log.dirs=/tmp/kafka-logs
    1. 启动kafka会从zookeeper下载配置到log目录
    2. 如果修改了server.properties可能因为配置与存储的配置不匹配导致启动失败,这时候可以删除这个目录
    3. 如果需要在同一台server上启动多个broker,这个路径必须不同
  • zookeeper.connect=localhost:2181
    1. zookeeper集群,以逗号隔开
  • listeners=PLAINTEXT://cvatap3d.nam.nsroot.net:9093
    1. broker的host:port

zookeeper与kafka

  • Kafka将Broker信息注册到zookeeper
    1. zookeeper会维护topic与broker的关系,选举Leader
    2. 监控partition leader存活性,发现Leader异常会重新选举Leader
    3. 当异常Broker恢复后,会在一段时间后重新分配Leader
    4. Broker从zookeeper获取集群中其它Broker信息
  • Consumer端将自己注册到zookeeper
    1. 用来获取broker列表
    2. 并和partition leader建立socket连接
    3. 在Consumer Group发生变化时进行rebalance
    4. Zookeeper管理consumer的offset跟踪当前消费的offset。
  • Producer端将自己注册到zookeeper
    1. 用来获取broker列表和分区状态,从而将消息发布到正确的Broker
    2. Zookeepr不管理producer