使用场景

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

    工作原理

  1. 生产者客户端应用程序产生消息:
    1. 客户端连接对象将消息包装到请求中发送到服务端
    2. 服务端的入口也有一个连接对象负责接收请求,并将消息以文件的形式存储起来
    3. 服务端返回响应结果给生产者客户端
  2. 消费者客户端应用程序消费消息:

    1. 客户端连接对象将消费信息也包装到请求中发送给服务端
    2. 服务端从文件存储系统中取出消息
    3. 服务端返回响应结果给消费者客户端
    4. 客户端将响应结果还原成消息并开始处理消息

      producer 生产者

  3. 代码执行 -> Interceptors 拦截器 -> Serializer 序列化器 -> Partitioner 分区器 -> DQuene 消息队列 -> sender 线程 -> borker

    1. RecordAccumulator 默认32MB ProducerBatch 默认16KB
    2. batch.size 当数据到达一定量时发送数据 默认16KB
    3. linger.ms 当到达一定时间发送数据 默认0ms
    4. 应答acks
      1. 0: 不需要等数据落盘应答
      2. 1:Leader收到数据后应答
      3. -1(all):Leader和ISR队列里面所有节点收齐数据后应答. -1和all等价
  4. 工作流程

    1. Producer直接发送消息到Broker上的Leader Partition
    2. Producer客户端自己控制着消息被推送到哪些Partition
    3. 随机分配、自定义分区算法等
    4. Batch推送提高效率

      consumer 消费者

  5. 消费者通过订阅消费消息
    offset的管理是基于消费组(group.id)的级别
    每个Partition只能由同一消费组内的一个Consumer来消费
    每个Consumer可以消费多个分区
    消费过的数据仍会保留在Kafka中
    消费者数量一般不超过分区数量
    消费模式
    队列:所有消费者在一个消费组内
    发布/订阅:所有消费者被分配到不同的消费组

    broker

    Kafka集群中每个Broker都可以响应Producer的请求
    哪些Broker是存活的?
    Topic的Leader Partition在哪?
    每个Broker充当Leader和Followers保持负载平衡
    Leader处理所有读写请求
    Followers被动复制Leader

ZooKeeper在Kafka中的作用

Broker注册并监控状态
/brokers/ids
Topic注册
/brokers/topics
生产者负载均衡
每个Broker启动时,都会完成Broker注册过程,生产者会通过该节点的变化来动态地感知到Broker服务器列表的变更
offset维护
Kafka早期版本使用ZooKeeper为每个消费者存储offset,由于ZooKeeper写入性能较差,从0.10版本后,Kafka使用自己的内部主题维护offset

  1. 启动后会向zookeeper中注册brokerId节点(/brokers/ids/[0,1,2])与topic节点(/brokers/topics),告诉zk其brokerid,集群情况下谁先注册谁为leader
  2. broker将的topic注册到zk中

    Kafka优化

    消息有序

  • Kafka保证在同一主题同一分区内有序
  • 如何确保基于主题全局有序

    • 一个主题一个分区
    • 生产者将消息按Key分组如(Table+PK),一个分组写入一个分区

      消息副本保证

  • request.required.acks

    • 生产者从不等待ack
    • 生产者等Leader写成功后返回
    • /all -生产者Leader和所有ISR中的Follower写成功后返回
  • min.insync.replicas

    • 该属性规定了最小的ISR数。当producer设置request.required.acks为all或-1时,指定副本(replicas)的最小数目,如果这个数目没有达到,producer会产生异常

      Kafka Producer API

  • 消息积累在Batch的缓冲区

  • 消息按分区批处理,正处于批处理级别重试中
  • 重试后,过期的批次被丢弃
  • Producer close/flush失败
  • 数据生产比交付快,导致BufferExhausedException

    FAQ

    为什么要使用 kafka,为什么要使用消息队列

    缓冲和削峰:上游数据时有突发流量,下游可能扛不住,或者下游没有足够多的机器来保证冗余,kafka在中间可以起到一个缓冲的作用,把消息暂存在kafka中,下游服务就可以按照自己的节奏进行慢慢处理。
    解耦和扩展性:项目开始的时候,并不能确定具体需求。消息队列可以作为一个接口层,解耦重要的业务流程。只需要遵守约定,针对数据编程即可获取扩展能力。
    冗余:可以采用一对多的方式,一个生产者发布消息,可以被多个订阅topic的服务消费到,供多个毫无关联的业务使用。
    健壮性:消息队列可以堆积请求,所以消费端业务即使短时间死掉,也不会影响主要业务的正常进行。
    异步通信:很多时候,用户不想也不需要立即处理消息。消息队列提供了异步处理机制,允许用户把一个消息放入队列,但并不立即处理它。想向队列中放入多少消息就放多少,然后在需要的时候再去处理它们。

    Kafka中的ISR、AR又代表什么?ISR的伸缩又指什么

    ISR:In-Sync Replicas 副本同步队列
    AR:Assigned Replicas 所有副本
    ISR是由leader维护,follower从leader同步数据有一些延迟(包括延迟时间replica.lag.time.max.ms和延迟条数replica.lag.max.messages两个维度, 当前最新的版本0.10.x中只支持replica.lag.time.max.ms这个维度),任意一个超过阈值都会把follower剔除出ISR, 存入OSR(Outof-Sync Replicas)列表,新加入的follower也会先存放在OSR中。AR=ISR+OSR。

    kafka中的broker 是干什么的

    broker 是消息的代理,Producers往Brokers里面的指定Topic中写消息,Consumers从Brokers里面拉取指定Topic的消息,然后进行业务处理,broker在中间起到一个代理保存消息的中转站。

kafka中的 zookeeper 起到什么作用,可以不用zookeeper么

zookeeper 是一个分布式的协调组件,早期版本的kafka用zk做meta信息存储,consumer的消费状态,group的管理以及 offset的值。考虑到zk本身的一些因素以及整个架构较大概率存在单点问题,新版本中逐渐弱化了zookeeper的作用。新的consumer使用了kafka内部的group coordination协议,也减少了对zookeeper的依赖,
但是broker依然依赖于ZK,zookeeper 在kafka中还用来选举controller 和 检测broker是否存活等等。

kafka follower如何与leader同步数据

Kafka的复制机制既不是完全的同步复制,也不是单纯的异步复制。完全同步复制要求All Alive Follower都复制完,这条消息才会被认为commit,这种复制方式极大的影响了吞吐率。而异步复制方式下,Follower异步的从Leader复制数据,数据只要被Leader写入log就被认为已经commit,这种情况下,如果leader挂掉,会丢失数据,kafka使用ISR的方式很好的均衡了确保数据不丢失以及吞吐率。Follower可以批量的从Leader复制数据,而且Leader充分利用磁盘顺序读以及send file(zero copy)机制,这样极大的提高复制性能,内部批量写磁盘,大幅减少了Follower与Leader的消息量差。

什么情况下一个 broker 会从 isr中踢出去

leader会维护一个与其基本保持同步的Replica列表,该列表称为ISR(in-sync Replica),每个Partition都会有一个ISR,而且是由leader动态维护 ,如果一个follower比一个leader落后太多,或者超过一定时间未发起数据复制请求,则leader将其重ISR中移除 。

kafka 为什么那么快

  • Cache Filesystem Cache PageCache缓存
  • 顺序写 由于现代的操作系统提供了预读和写技术,磁盘的顺序写大多数情况下比随机写内存还要快。
  • Zero-copy 零拷技术减少拷贝次数
  • Batching of Messages 批量量处理。合并小的请求,然后以流的方式进行交互,直顶网络上限。
  • Pull 拉模式 使用拉模式进行消息的获取消费,与消费端处理能力相符。

    kafka producer如何优化打入速度

  • 增加线程

  • 提高 batch.size
  • 增加更多 producer 实例
  • 增加 partition 数
  • 设置 acks=-1 时,如果延迟增大:可以增大 num.replica.fetchers(follower 同步数据的线程数)来调解;
  • 跨数据中心的传输:增加 socket 缓冲区设置以及 OS tcp 缓冲区设置。

kafka producer 打数据,ack 为 0, 1, -1 的时候代表啥, 设置 -1 的时候,什么情况下,leader 会认为一条消息 commit了

  1. 1(默认) 数据发送到Kafka后,经过leader成功接收消息的的确认,就算是发送成功了。在这种情况下,如果leader宕机了,则会丢失数据。
  2. 0 生产者将数据发送出去就不管了,不去等待任何返回。这种情况下数据传输效率最高,但是数据可靠性确是最低的。
  3. -1 producer需要等待ISR中的所有follower都确认接收到数据后才算一次发送完成,可靠性最高。当ISR中所有Replica都向Leader发送ACK时,leader才commit,这时候producer才能认为一个请求中的消息都commit了。

    kafka unclean 配置代表啥,会对 spark streaming 消费有什么影响

    unclean.leader.election.enable 为true的话,意味着非ISR集合的broker 也可以参与选举,这样有可能就会丢数据,spark streaming在消费过程中拿到的 end offset 会突然变小,导致 spark streaming job挂掉。如果unclean.leader.election.enable参数设置为true,就有可能发生数据丢失和数据不一致的情况,Kafka的可靠性就会降低;而如果unclean.leader.election.enable参数设置为false,Kafka的可用性就会降低。

    如果leader crash时,ISR为空怎么办

    kafka在Broker端提供了一个配置参数:unclean.leader.election,这个参数有两个值:
    true(默认):允许不同步副本成为leader,由于不同步副本的消息较为滞后,此时成为leader,可能会出现消息不一致的情况。
    false:不允许不同步副本成为leader,此时如果发生ISR列表为空,会一直等待旧leader恢复,降低了可用性。

    kafka的message格式是什么样的

    一个Kafka的Message由一个固定长度的header和一个变长的消息体body组成
    header部分由一个字节的magic(文件格式)和四个字节的CRC32(用于判断body消息体是否正常)构成。
    当magic的值为1的时候,会在magic和crc32之间多一个字节的数据:attributes(保存一些相关属性,
    比如是否压缩、压缩格式等等);如果magic的值为0,那么不存在attributes属性
    body是由N个字节构成的一个消息体,包含了具体的key/value消息

    kafka中consumer group 是什么概念

    同样是逻辑上的概念,是Kafka实现单播和广播两种消息模型的手段。同一个topic的数据,会广播给不同的group;同一个group中的worker,只有一个worker能拿到这个数据。换句话说,对于同一个topic,每个group都可以拿到同样的所有数据,但是数据进入group后只能被其中的一个worker消费。group内的worker可以使用多线程或多进程来实现,也可以将进程分散在多台机器上,worker的数量通常不超过partition的数量,且二者最好保持整数倍关系,因为Kafka在设计时假定了一个partition只能被一个worker消费(同一group内)。

    Kafka中的消息是否会丢失和重复消费?

    要确定Kafka的消息是否丢失或重复,从两个方面分析入手:消息发送和消息消费。

    1、消息发送

    Kafka消息发送有两种方式:同步(sync)和异步(async),默认是同步方式,可通过producer.type属性进行配置。Kafka通过配置request.required.acks属性来确认消息的生产:

  4. 0—-表示不进行消息接收是否成功的确认;

  5. 1—-表示当Leader接收成功时确认;
  6. -1—-表示Leader和Follower都接收成功时确认;

综上所述,有6种消息生产的情况,下面分情况来分析消息丢失的场景:
(1)acks=0,不和Kafka集群进行消息接收确认,则当网络异常、缓冲区满了等情况时,消息可能丢失
(2)acks=1、同步模式下,只有Leader确认接收成功后但挂掉了,副本没有同步,数据可能丢失

2、消息消费

Kafka消息消费有两个consumer接口,Low-level API和High-level API:

  1. Low-level API:消费者自己维护offset等值,可以实现对Kafka的完全控制;
  2. High-level API:封装了对parition和offset的管理,使用简单;

如果使用高级接口High-level API,可能存在一个问题就是当消息消费者从集群中把消息取出来、并提交了新的消息offset值后,还没来得及消费就挂掉了,那么下次再消费时之前没消费成功的消息就“诡异”的消失了;
解决办法
针对消息丢失:同步模式下,确认机制设置为-1,即让消息写入Leader和Follower之后再确认消息发送成功;异步模式下,为防止缓冲区满,可以在配置文件设置不限制阻塞超时时间,当缓冲区满时让生产者一直处于阻塞状态;
针对消息重复:将消息的唯一标识保存到外部介质中,每次消费时判断是否处理过即可。
消息重复消费及解决参考:https://www.javazhiyin.com/22910.html

为什么Kafka不支持读写分离?

在 Kafka 中,生产者写入消息、消费者读取消息的操作都是与 leader 副本进行交互的,从 而实现的是一种主写主读的生产消费模型。
Kafka 并不支持主写从读,因为主写从读有 2 个很明 显的缺点:

  • (1)数据一致性问题。数据从主节点转到从节点必然会有一个延时的时间窗口,这个时间 窗口会导致主从节点之间的数据不一致。某一时刻,在主节点和从节点中 A 数据的值都为 X, 之后将主节点中 A 的值修改为 Y,那么在这个变更通知到从节点之前,应用读取从节点中的 A 数据的值并不为最新的 Y,由此便产生了数据不一致的问题。
  • (2)延时问题。类似 Redis 这种组件,数据从写入主节点到同步至从节点中的过程需要经 历网络→主节点内存→网络→从节点内存这几个阶段,整个过程会耗费一定的时间。而在 Kafka 中,主从同步会比 Redis 更加耗时,它需要经历网络→主节点内存→主节点磁盘→网络→从节 点内存→从节点磁盘这几个阶段。对延时敏感的应用而言,主写从读的功能并不太适用。

    Kafka中是怎么体现消息顺序性的?

    kafka每个partition中的消息在写入时都是有序的,消费时,每个partition只能被每一个group中的一个消费者消费,保证了消费时也是有序的。
    整个topic不保证有序。如果为了保证topic整个有序,那么将partition调整为1.

    消费者提交消费位移时提交的是当前消费到的最新消息的offset还是offset+1?

    offset+1

    kafka如何实现延迟队列?

    Kafka并没有使用JDK自带的Timer或者DelayQueue来实现延迟的功能,而是基于时间轮自定义了一个用于实现延迟功能的定时器(SystemTimer)。JDK的Timer和DelayQueue插入和删除操作的平均时间复杂度为O(nlog(n)),并不能满足Kafka的高性能要求,而基于时间轮可以将插入和删除操作的时间复杂度都降为O(1)。时间轮的应用并非Kafka独有,其应用场景还有很多,在Netty、Akka、Quartz、Zookeeper等组件中都存在时间轮的踪影。
    底层使用数组实现,数组中的每个元素可以存放一个TimerTaskList对象。TimerTaskList是一个环形双向链表,在其中的链表项TimerTaskEntry中封装了真正的定时任务TimerTask.
    Kafka中到底是怎么推进时间的呢?Kafka中的定时器借助了JDK中的DelayQueue来协助推进时间轮。具体做法是对于每个使用到的TimerTaskList都会加入到DelayQueue中。Kafka中的TimingWheel专门用来执行插入和删除TimerTaskEntry的操作,而DelayQueue专门负责时间推进的任务。再试想一下,DelayQueue中的第一个超时任务列表的expiration为200ms,第二个超时任务为840ms,这里获取DelayQueue的队头只需要O(1)的时间复杂度。如果采用每秒定时推进,那么获取到第一个超时的任务列表时执行的200次推进中有199次属于“空推进”,而获取到第二个超时任务时有需要执行639次“空推进”,这样会无故空耗机器的性能资源,这里采用DelayQueue来辅助以少量空间换时间,从而做到了“精准推进”。Kafka中的定时器真可谓是“知人善用”,用TimingWheel做最擅长的任务添加和删除操作,而用DelayQueue做最擅长的时间推进工作,相辅相成。
    参考:https://blog.csdn.net/u013256816/article/details/80697456

    Kafka中的事务是怎么实现的?

    参考:https://blog.csdn.net/u013256816/article/details/89135417

    Kafka中有那些地方需要选举?这些地方的选举策略又有哪些?

    https://blog.csdn.net/yanshu2012/article/details/54894629