Kakfa基础

目录

写在前面

写了什么：Kakfa作为一个分布式消息系统设计过程中考虑的点，碎片知识体系化，细节不展开以后填坑
推荐阅读：有Kafka使用基础，帮助形成完整知识体系

kafka架构

start

bin/zookeeper-server-start.sh config/zookeeper.properties
bin/kafka-server-start.sh config/server.properties
bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic test
bin/kafka-topics.sh --list --bootstrap-server localhost:9092
bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic test
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning

broker

需要多少broker

  • 需要多少磁盘空间来保留数据,以及单个broker有多少空间可用,比如集群需要10TB,单个broker 2TB,就需要5个broker,如果开启了数据复制就要更多broker
  • 集群处理请求的能力,这通常与网络接口处理客户端流量的能力有关

producer

发送消息到kafka

  • 发送并忘记
producer.send(record);
  • 同步发送消息
producer.send(record).get();
  • 异步发送消息
producer.send(record,new DemoProducerCallBack());
privateclass DemoProducerCallBack implements Callback{
    @Override
    public void onCompletion(RecordMetadata metadata, Exception exception) {
        if(exception != null){
            exception.printStackTrace();
        }
        System.out.println("#####");
    }
}

生产者的配置

  • acks
    • acks = 0 高吞吐量,不需要服务器响应,容易丢消息
    • acks = 1 集群首领节点收到消息,仍然会丢消息, 同步增加延迟,异步高吞吐
    • acks = all 所有复制节点收到消息,最可靠,吞吐量最低
  • buffer.memory
    • buffer.memory = 33554432 生产者内存缓存区大小,如果发送消息速度超过发送到服务器速度,缓存区满空间不足,send会抛出异常或者阻塞,通过max.block.ms = 60000设置阻塞事件
  • retries
    • retries = 3 retry.backoff.ms = 100 出现异常,每隔100ms重试,重试3次
  • batch.size
    • batch.size = 16384 , 消息按照批次发送减少网络开销
  • linger.ms
    • linger.ms = 10 , 发送批次之前等待消息的最大时间,这两个参数搭配使用
  • max.request.size
    • max.request.size = 1048576 发送单个消息的最大值
  • max.in.flight.requests.per.connnection
    • max.in.flight.requests.per.connection = 5 指定了生产者在收到服务器响应之前可以发送多少个消息,对消息顺序敏感,设置成1,即使重试也不会乱序

consumer

消费者和消费者群组

  • 单播与广播

消费者群组和分区再均衡
分配策略：

  1. RangeAssignor
  2. RoundRobinAssignor
  3. StickyAssignor
  4. 自定义分区策略

创建kafka消费者

  • 轮询
    • 拉数据,发送心跳, 再均衡, 提交偏移量

消费者配置

  • fetch.min.bytes
    • fetch.min.bytes = 1, 消费者从服务器获取记录的最小字节数
  • fetch.max.wait.ms
    • 等待{fetch.min.bytes}数据到达的最大时间
  • max.partition.fetch.bytes
    • 每个分区返回给消费者的最大字节数,默认1MB
  • session.timeout.ms
    • 消费者不发送心跳的最大时间,默认3s,如果超过改时间没有发送,会被认为死亡,协调器再均衡
  • auto.offset.reset
    • 消费者读取没有偏移量的分区或者偏移量无效的情况下如果处理,默认lastest,从最新开始读(消费者启动后的记录),earliest从分区起始位置读
  • enable.auto.commit
    • 消费者是否自动提交偏移量,可能出现重复数据或者数据丢失,默认是true,auto.commit.interval.ms后自动提交,false是手动提交
  • partition.assignment.strategy
    • 分区分配给群组消费者的分配策略,有两种range:将主题的若干分区分配给消费者,RoundRobin把主题的所有分区逐个分配给消费者,默认是range

线程安全
一个线程无法运行多个消费者,也无法让多个线程共享一个消费者,一个消费者使用一个线程,解决方案是:
1. 将消费者的逻辑封装在自己的对象里,然后使用Executor去执行,需要考虑消息丢失(类似滑动窗口提交偏移量)和消息顺序(无法保证)
2. 创建多个KafkaConsumer实例,缺点是网络开销比较大

提交和偏移量
在旧的消费者客户端中,消费位移存储在zookeeper中的,新的消费者向_consumer_offset的特殊topic发送消息用来记录每个分区的偏移量,当触发再均衡,分区被分配到新的消费者,消费者通过该topic上记录的分区偏移量开始读取

  • 自动提交
    • 按照auto.commit.interval.ms时间间隔来提交偏移量,具体做法是在每次poll的时候检查是否该提交偏移量了,如果是,提交上一次轮询返回的偏移量,修改提交时间更频繁的提交偏移量,减小重复的时间窗,但无法避免,至少重复一个批次的消息
  • 同步提交
    • 设置enable.auto.commit false, commitSync()提交poll()返回的最新偏移量,发生再均衡,最近一批消息到发生再均衡之间的所有消息会被重复处理
  • 异步提交
    • commitAsync()提高commitSync()的吞吐量,异步提交一般不会重试,否则会出现提交了3000,前面异步重试提交2000,可能导致重复处理,解决方法是回调设置一个单调递增的序列号和提交偏移量对比,如果相等说明没有新的提交,可以进行重试
  • 提交特定的偏移量
    • 上述是一个批次提交消息,如果想要更频繁的提交消息,自己维护一个Map<TopicPartition,OffsetAndMetadata> currentOffsets, 在一个批次里面设定处理消费1000record后commitSync(currentOffsets),这样重复数据最多重复1000record
  • 再均衡监听器
    • 实现ConsumerRebalanceListener重写onPartitionsRevoked和onPartitionAssigned方法,实现在消费者失去分区所有权停止读取消息和另一个消费者获得该分区准备读取消息之前的回调
      • 消费者维护一个分区/偏移量的map,在onPartitionsRevoked中提交到_consumer_offset
      • 消费者维护一个分区/偏移量的map,在onPartitionsRevoked中存储到DB,在onPartitionAssigned通过seek查到偏移量开始消费
  • 从特定偏移量读取
    • 将处理数据和提交偏移量放在本地事务写入数据库,实现再均衡监听器,在onPartitionAssigned中便利partition调用seek从数据库读取偏移量

如何退出

  1. 注册安全回调wakeUp方法(修改AtomicBoolean wakeup值)到shutdownHook
  2. poll会校验wakeUp值,如果为true,抛出WakeUpException退出
  3. 在finally里面调用consumer.close完成清理工作
        Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
            @Override
            public void run() {
                consumer.wakeup();
            }
        }));

partition

org.apache.kafka.clients.producer.internals.DefaultPartitioner#partition
路由策略:

  • 指定partition
  • 指定key,分区不变的情况下同一key固定散列到固定槽,分区变化键和分区的对应关系会变化
  • key为null,Round Robin(轮询)分配分区
  • 对于热点key可以自定义分区器(实现Partitioner接口),将热点key散列到固定分区，其他key散列到其他分区

设定策略:
根据partition的吞吐量和消费者消费能力,设定合适的partition,预估topic吞吐量/消费者吞吐量 = 分区数,如果无法预估,单个分区25G内较好

Interceptor

public class KafkaProducerInterceptor implements ProducerInterceptor<String, String> {
    //producer.send的时候调用
    public ProducerRecord<String, String> onSend(ProducerRecord<String, String> producerRecord) {
    ....
    }
    //异步IO线程在接受broker响应的时候调用,避免耗时调用
    public void onAcknowledgement(RecordMetadata recordMetadata, Exception e) {
    ...
    }
}

深入kafka

集群成员关系:
每个broker都有一个唯一标示,在broker启动的时候,创建临时节点将id注册在zookeeper上(路径是/brokers/ids),其他kafka订阅zk,实现broker信息的获取

控制器

  • 控制器选举

    1. 去zk创建/controller临时结点,{"version":1,"brokerUid":0,"timestamp":1529210278988}
    2. 去zk创建/controller_epoch持久结点,控制器纪元
    3. 充当控制器的broker会注册监听器到相应的结点上,本地通过事件队列线程模型通知给其他broker
    4. 其他broker会监听/controller变更,当控制器broker下线,开始新的选举

  • 分区leader选举

复制:

  • 首领副本
    • 每个分区都有一个首领副本,为了保证一致性,所有生产者消费者请求都会经过这个副本
  • 跟随者副本
    • 不处理请求,从首领那里复制消息,保持与首领一致的状态,如果首领崩溃,其中一个同步的副本提升为新首领,跟随者类似消费者给leader发送获取数据的请求,请求消息中包含想要获取的消息的偏移量,leader因此得知follower的同步进度,如果长时间没有同步或者没有最新数据或者没有给zk发送心跳被认为是不同步的,不会被选举为leader
  • 非同步副本
    • 跟随者副本的一种状态,由于消费者消费完全同步的消息,生产者的ack也依赖是否完全同步,所以非同步副本将不纳入是否同步的分区之内,非同步副本不会对性能产生任何影响,但是发生宕机时丢失数据的风险也就越大
  • 副本设定策略
    通过Replication factor来指定,至少设置成2,建议是3,最高设置为4,太少broker宕机消息可靠性不能保证,太多消息延时较大花费更多的磁盘空间,也是可靠性和性能之间的权衡
    • 以3为起始（当然至少需要有3个brokers，同时也不建议一个Kafka 集群中节点数少于3个节点）
    • 如果replication 性能成为了瓶颈或是一个issue，则建议使用一个性能更好的broker，而不是降低RF的数目
    • 永远不要在生产环境中设置RF为1

处理请求:

  • 生产请求
    • 生产者发送的请求,它包含客户端要写入broker的消息,根据ack确定是否发送成功
  • 获取请求
    • 在消费者或者副本需要从broker读取消息时发送的请求,只能读取已经同步副本的消息
  • "非分区首领"错误
  • 元数据请求
  • 元数据刷新

分区个数:
并不是越多越好,在达到某个阈值后会下降,分区数是有上限的,分区的增加保存在zk,producer,consumer的分区信息会增加,同时会延迟主备同步,宕机后的leader选举

文件管理:
分区分成若干片段,每个片段包含1GB或一周的数据,以较小的那个为准,在broker给分区写数据时,如果达到片段上线,就关闭当前文件,并打开一个新文件

文件格式:
发送者发送的消息和消费者接受的消息格式是一样的,这样可以通过零复制给消费者发消息,同时避免了消息的解压以及再压缩,如果按照批次发送,同一批次的消息也会被压缩在一起,被当作包装消息"进行发送",

日志索引

核心文件.log(存储消息), .index(索引文件), .timeindex(时间索引文件)

  • 偏移量索引
    1. 通过ConcurrentSkipListMap定位.index文件
    2. 二分查找找到position物理偏移量
    3. 去.log文件根据log查看消息
  • 时间戳索引
    1. 顺序查找.timeIndex
    2. 二分查找得到relativeOffset
    3. 上述三个步骤

日志清理

  1. 基于时间
  2. 基于日志大小
  3. 基于日志起始偏移量

磁盘存储

  1. 顺序写页缓存
  2. 零拷贝
    1. sendFile
    2. FileChannel.transferTo

可靠的数据传输

  • kafka保证分区消息的顺序
  • 当有消息被写入分区的所有同步副本时,它才被认为是"已提交"的,
  • 只要还有一个副本活跃,已提交的消息就不会丢失
  • 消费者只能读取已提交的消息

不完全的首领选举
当分区首领宕机,这个时候没有同步副本,只有不同步副本,这个是否让不同步的副本来成为首领吗,通过下述参数设置
unclean.leader.election.enable

  • true, 允许,会有数据丢失的风险,但是不会不可用
  • false, 不允许, 等待宕机的首领重新恢复,会短暂不可用,例如银行业务

最少同步副本
在topic和broker级别上可以设定,分区同步副本的最小值,min.insync.replicas和acks=all搭配使用,如果当前同步副本小于该值,broker不会接受消息这个时候变成只读的

在可靠的系统里使用生产者

  • 发送确认
    • acks = 0
    • acks = 1
    • acks = all
  • 配置生产者的重试参数
    • 生产者可以自动处理的错误
    • 开发者手动处理的错误

在可靠的系统里使用生产者

  • 消费者的可靠配置
    • enable.auto.commit = false, 决定提交方式
  • Kafka支持仅一次传递
    • 消费者开启幂等配置enable.idempotence=true
    • kafka事务属性
  • 长时间处理
    • 消息处理有外部依赖可能耗时很长,为了避免分区rebalance,提高消费效率，可以使用线程池来处理数据,会有消息丢失风险，参考滑动窗口设计保证消息不丢，可能消息重复

mq使用场景

  • 解耦
    基于发布订阅，业务方快速接入，系统耦合降低，改造成本低

  • 异步
    发短信异步

  • 削峰
    高并发场景通过异步队列削峰

常见问题

  • 消息重复
    出现场景:网络抖动,异常重试,机器宕机,本质是做下幂等处理
    弱校验: 加上redis分布式锁
    强校验: 可以落地一个同步日志表,建立唯一索引,就算插入也会重复失败

  • 消息丢失

  • 消息顺序

【推广】 免费学中医，健康全家人
原文地址：https://www.cnblogs.com/kundeg/p/13456094.html