kafka

1.kafka简介

          kafka消费模式：主要有2种，分别是一对一的消费和一对多的消费。

                 一对一消费：点对点通信，一个发送，一个接收。消息发送者将消息发送至队列中，通知消费者从队列中拉去数据进行消费，消费完毕后，队列中消息删除。1条消息只能被1个消费者消费

                 一对多消费：也成为发布/订阅模式，利用topic存储消息，消息生产者将消息发布到topic中，多个消费者监听此topic，多个消费者从topic中消费信息。

                                      消费者消费数据之后，数据不会被删除。kafka会默认保留一段时间，然后再删除。

          kafka基础架构：

                 Producer:消息生产者，向kafka中发布信息的角色

                 Cousumer:消息消费者，从kafka中拉取信息消费的客户端

                 Consumer Group:消费者组，一组中存在多个消费者，消费者消费broker中当前topic的不同分区的消息，

                               消费者组之间互不影响，所有消费者都属于某个消费者组，消费者组是逻辑上的一个订阅者。

                               某一分区的消费只能够一个消费者组中的一个消费者所消费。

                 Broker:经纪人，一台kafka服务器就是一个broker,一个集群由多个broker组成，一个broker可容纳topic

                 Topic:主题，理解为队列，生产者和消费者都是面向topic

                 Partition:分区，为了实现扩展性，一个很大的topic可以分布在很多个broker上，一个topic中包含多个partition,每个partition是一个有序的队列(可保证分区有序，无法保证全局有序)

                 Replica:副本Replication,为保障集群中某个节点发生故障，节点上的Partition数据不丢失，kafka可以正常工作，提供了副本机制

                               一个topic的每个分区有若干个副本，一个Leader和若干个Follower

                 Leader:每个分区多个副本的主角色，生产者发送数据的对象及消费者消费数据的对象都是Leader

                 Follower:每个分区多个副本的从角色，实时从Leader中同步数据，保持和Leader同步，Leader发生故障，某个Follower会称为新的Leader，此时会对齐消息的偏移量

          kafka安装使用

                 安装命令示例：

                        拉去kafka镜像：docker pull wurstmeister/kafka;

                        拉去zookeeper镜像：docker pull wurstmeister/zookeeper;

                        启动zookeeper(先启动zookeeper是为了kafka有地方注册消息):docker run -it --name zookeepre -p 12181:2181 -d wurstmeister/zookeeper:latest

                        启动3台kafka，端口都是映射到9092

                        第一台：

                        docker run -it --name kafka01 -p 19092:9092 -d -e KAFKA_BROKER_ID=0 -e KAFKA_ZOOKEEPER_CONNECT=ip:12181 -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://ip:19092 -e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 wurstmeister/kafka:latest

                        第二台：

                        docker run -it --name kafka02 -p 19093:9092 -d -e KAFKA_BROKER_ID=1 -e KAFKA_ZOOKEEPER_CONNECT=ip:12181 -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://ip:19093 -e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 wurstmeister/kafka:latest

                        第三台：

                        docker run -it --name kafka03 -p 19094:9092 -d -e KAFKA_BROKER_ID=2 -e KAFKA_ZOOKEEPER_CONNECT=ip:12181 -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://ip:19094 -e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 wurstmeister/kafka:latest

                        注意：1台服务器同一个kafka的端口可启动多个broker，组成集群，但此种方式不常见，通常都是1台kafka对应1个broker，除非测试或资源优先，否则不建议这样做。

                 具体命令：

                        创建topic名称为first,3个分区，1个副本：./kafka-topics.sh --zookeeper ip:12181 --create --topic first --replication-factor 1 --partitions 3

                        查看kafka中first的topic信息：./kafka-topics.sh --zookeeper ip:12181 --describe --topic first

                        调用生产者生产消息：./kafka-console-producer.sh --broker-list ip:19092,ip:19093,ip:19094 --topic first

                        调用消费者消费消息：./kafka-console-consumer.sh --bootstrap-server ip:19092,ip:19093,ip:19094 --topic first --from-beginning

                                                           --from-beginning 表示读取全部消息

                        删除topic:./kafka-topic.sh --zookeeper ip:12181 --delete --topic first;前提需要设置config/server.properties中delete.topic.enable=true

                        修改分区数：./kafka-topics.sh --zookeeper ip:12181 --alter --topic first --partitions 3

                                             first表示的是主题名，3表示修改之后的分区数量

2.kafka高级

          工作流程:

                 kafka中的消息是以topic进行分类的，Producer生产消息，Consumer消费消息，都是面向topic的

                 topic是逻辑上的实现，partition是物理上的概念，每个partition对应着一个log文件，此Log文件中存储的就是producer生产的数据，topic=N*partition,partiton=log

                 producer生产的数据会不断追加到Log的末端，且每条数据都有自己的offset,消费者组中的每个consumer，都会实时记录自己消费到那个offser,以便出错恢复，可以从上次的位置继续进行消费。流程：Producer=>Topic（Long With offset）=>Cousumer

          文件存储

                 通过本地落盘的方式存储的，主要是通过相应的log与index等文件保存具体的消息文件

                 一个topic包含多个partition,一个partition包含多个segment,一个segment对应一个.log文件和.index文件

                 生产者不断向log文件中追加消息文件，为了防止文件过大，导致效率低下，kafka的log文件以1G作为临界点，当.log文件大小超过1G时，会创建1个新的.log文件，同时为了快速定位大文件中消息位置，kafka采取了分片和索引的机制来加速定位。

                 在kafka存储log文件的地方，会存在消费的偏移量和具体的分区信息，分区信息主要包括.log和.dms组成，分区目的是为了备份。

                 快速定位数据：.index存储的消息是offset+真实的起始偏移量，.log存储的是真实的数据

          生产者分区策略

                 分区的原因：方便在集群中扩展，每个分区通过调整可以适应所在的机器，而每一个topic又包含多个分区，因此整个集群可以适应适合的数据；可提高并发，以分区为单位进行读写。类似多路

                 分区原则：

                        指定是第几分区的情况下，直接将指明的值作为分区名称；

                        未指明分区，但是存在值Key,此时将Key的hash值与partition总数取余得到partition的值;

                        值与partition均没有，第一次随机生成1个整数，后面每次调用在这个基础上自增，将这个值与topic可用的partition总数取余得到partition值，即round-robin算法，

          2.4 生产者ISR：

                 为保证producer发送的消息能够可靠发送到每个topic中，topic中的每个partition在收到producer发送的数据后，都需要向producer发送acknowledgement，如果producer收到ack后就进行下一轮发送，否则重新发送数据

                 发送ack时机：确保leader和follower同步完成，再发送ack，可以确保leader挂掉之后，可以选出新的。主要是确保follower数据不丢失

                 follower同步完成多少才发送ack：半数以上的follower同步完成，即可发送；全部follower同步完成，才会发送

                 半数以上follower同步完成后发送ack：优点是延迟低；缺点是选举新的leader，如果要容忍n个节点的故障，需要2n+1个follower副本，容错率低，为1/2

                 全数follower同步完成后发送ack:优点是容错率高，缺点是延迟高，需要全部副本同步完成才可。

                 kafka选择的是全数同步后发送的方式，因为容错率更有优势，同时对于分区数据而言，每个分区都有大量数据，半数同步会造成大量数据冗余。虽然全部同步延迟高，但是网络延迟对kafka影响较小。

                 ISR(同步副本集)：采用全数副本同步完成后发送，如果有follower故障迟迟完成不了同步，因此会导致迟迟无法发送ack。对此leader中维护了一个动态的ISR，如果follower迟迟没有同步完成，会将此follower从ISR踢出

          2.5 生产者ACK机制

                 对于某些不重要的数据，可容忍少量数据丢失，并非必须全部等待所有副本同步后才可发送，因此kafka提供了3种ack可靠性级别，从而可根据可靠性和延迟的要求进行合适的配置

                 ack参数配置：

                        0:producer不等待broker的ack，延迟最低，broker还未将数据写入磁盘就返回，broker故障事可能丢失数据

                        1：producer等待broker的ack,partition的leader落盘成功后返回ack,如果follower同步成功之前leader故障，那么数据丢失

                        -1(all)：partition中的leader和follower全部落盘成功后返回ack，如果在follower同步完成后，broker发送ack之前，leader发生故障，会造成数据重复

          2.6 数据一致性问题

                 LEO(Log And OffSet):每个副本的最后一个offset

                 HW(High Watermark):高水位，指代消费者能看到的最大的Offset,所有副本（ISR队列）中最小的LEO

                 follower发生故障：之后会被临时提出ISR，等待改follower恢复后，follower会读取本地磁盘上的HW，并将log文件高于HW的部分截掉，从HW开始向leader同步，等待改follower的LEO大于等于改partition的HW，即follower追上leader，就可以重新加入ISR了

                 leader发生故障：之后从follower中选择一个新的Leader,为了保证多个副本数据一致性，其余副本会将各自Log文件中高于HW的部分截掉，然后从新的leader重新读取数据

                 以上只能保证数据的一致性，无法保证数据不丢失或不重复

          2.7 ExactlyOnce

                 服务器ACK级别设置为-1，可保证producer和broker之间数据不会丢失，即至少一次语义(at least once)。ACK级别设置为0，可保证生产者每条消息只会发送一次，即至多一次。(at most once)

                 at least once可保证数据不丢失，但是不能保证数据不重复；at most once可保证数据不重复，但无法保证数据不丢失。对于重要数据，既不能丢失也不能重复，exactlyonce即精确一次。

                 0.11版本的kafka，引入了幂等性，即生产者无论发送多少次重复数据，server端只会持久化1条数据；集合at least once 构成了extractlyonce

                 kafka的幂等性实际上是将数据去重放在了数据上有来做，开启幂等性的producer会被分配一个PID,发往同一个分区的Sequence Numer，而broker端会对<PID,paration,seqnumer>进行缓存。

                 但是PID会在重启后会变化，同时不同的partitionyeyou 不同的主键，所以幂等性无法保证跨分区跨会话的exectly once

3.消费者分区分配策略

                 消费方式：consumer采用pull的方式从broker中获取数据，如果采用Push的方式推送数据，很难适应消费速率不同的消费者，因为broker目标是尽可能以最大速度将数据传递，但这样容易导致consumer来不及处理消息，典型的表现就是拒绝服务及网络堵塞，而pull可让消费者根据自己的消费处理能力以适当的速度消费消息。pull模式的不足在于如果kafka没有数据，消费者可能会陷入循环中(因为消费者类似监听状态获取数据消费的)，针对这点,kafka的consumer在消费数据事会传入一个timeout,当前如果没有数据可消费，consumer会等待一会之后再返回

                 3.1 分区分配策略：

                        一个consumer group 中有多个consumer,一个topic中包含多个partition，必然涉及到partition的分配问题，即确定那个partition由那个consumer消费的问题

                        kafka两种分配策略：

                               round-robin循环：采用轮询的方式分配所有分区，此方式会导致每个consumer所承载的分区数不一致，从而导致各个consumer压力不均

                               range:首先计算各个consumer所能承载的分区数量，然后将指定数量的分区分配给该consumer

                 3.2 消费者offset的存储

                        consumer在消费过程中可能出现断点宕机等故障，consumer需要记录消费到了那个offset,以便恢复之后可以继续消费

                        kafka0.9版本之前，consumer将offset保存在zookeeper中，0.9版本之后，consumer默认将Offset存储在内置的topic中，为_consumer_offsets

                 3.3 消费者组案例

                        选定一个消费者组，1条消息只会被一个组中的一个消费者消费，只有ctrl+c退出其中一个消费者，另外的消费者才会继续消费

4.高效读写&Zookeeper的作用

          4.1 kafka的高效读写

                 顺序写磁盘：kafka的producer生产数据，是通过写在Log文件末端，采用追加的方式进行写入。同样的磁盘，顺序写的方式，速度大约600M/s,随机写的方式只有200k/s.顺序写之所以快，是因为省区了大量磁头寻址的时间。

                 零复制技术：将磁盘文件中的数据复制到页面缓存中一次，然后将数据从页面缓存直接发送到网络中，避免了重复复制的操作。例如传统方式下，10个消费者，推送给每个消费者需要4步，合计40步骤。但如果使用零拷贝技术，只需要11步即可。为将数据推送到页面缓存和每个消费者从页面缓存中读取。

          4.2 kafka中zookeeper的作用

                 kafka集群中有一个broker会被选举为controller,负责管理集群的上下线、所有的topic分区副本分配和leader的选举等工作。controller的工作是依赖于zookeeper的

5.事务

          从0.11版本引入了事务的支持，可以保证在exactly once的基础上，生产和消费可以跨分区的会话，要么全部成功，要么全部失败

          5.1 producer事务

                 引入了全局事务id，并将producer_id和事务id进行绑定，这样即便重启之后也可根据transaction_id获取producer_id

                 为了管理transaction,还引入了transaction coordinator,producer是通过和transaction coordinator交互获取transaction——id对应的任务状态；transaction coordinator还将事务写入内部的topic中，这样即便服务重启，由于事务得到了保存，也可以继续运行。

          5.2 consumer事务

                 consumer的事务相对producer较弱，无法确保commit的消息被精准消费，因为consumer可通过Offset访问任意消息，而且不同的segment file生命周期不同，同一事务的消息可能出现重启后被删除的情况

6.API生产者流程

          producer发送消息采用的是异步发送，发送过程中涉及到了2个线程，分别是main线程和sender线程，以及一个线程变量recordaccumulator，main线程将消息发送给recordaccumulator,send线程不断重recordaccumulator拉去信息发送到kafka broker中

          异步发送带回调函数的生产者：回调函数在producer收到ack时调用，为异步调用，此含税有2个参数，分别为recordmetadata和exception，如果exception为null,发送成功，否则发送失败。消息失败后会启动重试机制，但要在回调函数中手动重试。

          生产者分区策略测试：ProducerRecord中的partition参数即为指定的分区（每个分区都有各自的编号，此参数是分区编号）

          自定义分区构造：实现Partitioner,重写其方法即可

          Producer发送消息默认是异步方式，如果要同步发送消息，则send之后，需要获取到具体的future的值，通过调用future.get()方法可以暂时阻塞，以达到同步发送的目的

7.APi消费者

          7.1 简单消费者

                 消费者开启了自动提交offset的功能：offset.auto.commit=true

          7.2 消费者重置offset

                 kafka中数据是持久化的，不用担心数据丢失问题。但是consumer在消费过程中可能出现宕机等状况，因为consumer需要记录自己消费的Offset位置，以便服务恢复后可继续消费

                 可通过auto.offset_reset来进行控制，有以下4种取值

                 earliest:重置offset到最早的位置

                 latest:重置offset到最新的位置，默认值

                 none:在消费者组种找不到前一个则抛出异常

                 anything else:抛出异常给消费者

          7.3 消费者保存Offset读取问题：enable.auto.commit=true 即自动提交offset,默认自动提交

          7.4 消费者手动提交offset

                 自动提交很便捷，但是难以把我Offset提交的时机，因此kafka提供了2种手动提交offset的api

                 commitSync:同步提交

                 commitAsync:异步提交

                 相同点：以上两种方式都会将本次poll拉去的数据中最高的偏移量提交

                 不同点：commitSync阻塞当前线程，持续到提交成功，失败会自动重试；commitAsync则没有失败重试机制，有可能提交失败，异步提交多了一个提交的回调函数

          7.5 数据漏消费和数据重复消费

                 先提交offset后消费，可能导致数据漏消费

                 先消费后提交offset,可能导致数据重复消费

          7.6 自定义存储Offset

                 分区重新分配的过程称为Rebalance.

                 消费者发生rebalance之后，每个消费者的分区会发生变化，因此消费者需要先获取到重新分配的分区，并且定位到每个分区的Offset位置继续进行消费。

                 实现思路：记录下需要提交的offset,利用rebalance分区监听器监听rebalance事件，一旦发生rebalance，先将Offset提交，分区之后找到最新的offset重新消费即可

8.自定义拦截器

          producer拦截器是kafka0.10引入的，主要用于clients端的定制化控制逻辑，对producer而言，拦截器可在发送消息之前和回调逻辑之前对消息做一些定制化需求。

          拦截器实现的接口是ProducerInterceptir,主要以下方法

                 configure(Map<String, ?> configs)：获取配置信息和初始化数据时调用

                 onSend(ProducerRecord record)：该方法封装在KafkaProducer.send()方法中，运行在用户主线程中，Producer确保在消息被序列化之前及计算分区前调用该方法，并且通常都是在Producer回调逻辑出发之前。

                 onAcknowledgement(RecordMetadata metadata, Exception exception)：onAcknowledgement运行在Producer的IO线程中，因此不要再该方法中放入很重的逻辑，否则会拖慢Producer的消息发送效率。

                 close()：关闭inteceptor，主要用于执行资源清理工作

                 zinterceptor可能运行在多个线程中，因此运行事件需要自行控制线程安全。另外如果指定了多个拦截器，则拦截器按照指定顺序调用，并仅仅只是捕获每个interceptor可能抛出的异常记录在错误日志中而非向上传递

                 自定义中加入时间戳拦截器、自定义消息发送统计拦截器、在consumerProducer加入拦截器

                 注意：kafka拦截器的close是首位的，一定要调用producer的close方法，否则拦截器的close方法不会被调用。

9.kafka监管eagle

          eagle:开源的可视化和管理软件，为kafka集群提供了将kafka集群数据转为图形和可视化的工具

10.常见题目

          kafka中的ISR、OSR、AR分别表示什么？

                 ISR(InSyncRecli):速度和leader低于10s的follower组成的集合

                 OSR(OutSyncRepli):速度和leader相差大于10s的follow组成的集合

                 Ar（AllRecli）:所有分区

          kafka中的HW,LEO等分别代表什么每个分区中最大的OFFSER

                 HW(High wATER):高水位，根据同一分区中最低的LEO决定

                 LEO:每个分区最大的offset