Kafka

高性能的消息中间件，在大数据的业务场景下性能比较好，kafka本身不维护消息位点，而是交由Consumer来维护，消息可以重复消费，并且内部使用了零拷贝技术，性能比较好Broker持久化消息时采用了MMAP的技术，Consumer拉取消息时使用的sendfile技术

Kafka是最初由Linkedin公司开发，是一个分布式、支持分区(parition)、多副本的(replica)，基于Zookeeper协调的分布式消息系统，它最大的特性就是可以实时地处理大量数据以满足各种需求场景比如:基于Hadoop的批处理系统、低延迟的实时系统、Storm/Spark流式处理引擎，web/nginx日志、访问日志，消息服务等等，用scala语言编写

吞吐量在10w~100w

概述

模型设计图

一个parition只能交给一个consumer消费，因为交给多个consumer让其进行poll拉取消息进行消费，会引起重复消费的问题

服务端(broker)和客户端(producer、consumer)之间通信通过TCP协议来完成

同时未发布和订阅提供高吞吐量，Kafka的设计目标是以时间复杂度O(1)的方式提供消息持久化能力的，即使对TB级别以上数据也能保证常数时间的访问性能，即使在非常廉价的商用机器上也能做到单机支持每秒100K条消息的传输

消费消息采用Pull模式，消息被处理的状态是在Consumer端维护的，而不是由服务端维护，Broker无状态，Consumer自己保存offset，这样消息的消费速度直接和消费者有关，broker只是存储消息

特点

Kafka根据topic对消息进行归类，发布到Kafka集群的每条消息都需要指定一个topic

Topic

物理上的概念，一个topic可以分为多个partition，每个partition的内部时有序的

Parition

消息中间件处理节点，一个Kafka节点就是一个broker，一个或者多个Broker可以组成一个Kafka集群

Broker

每个Consumer属于一个特定的ConsumerGroup，一条消息可以被多个不同的ConsumerGroup消费到那时一个ConsumerGroup中只能有一个Consumer能够消费该消息

ConsumerGroup

消息消费者，从Broker读取消息的客户端

Consumer

消息生产者，向Broker发送消息的客户端

Producer

核心组件

在Kafka集群中会有一个或者多个broker，其中有一个broker会被选举为控制器(Kafka Controller)，它负责管理整个集群中所有分区和副本的状态1.当某个分区的leader副本出现故障时，由控制器负责为该分区选举出新的leader副本2.当检测到某个分区的ISR(In-Sync-Replica)集合发生变化时，由控制器负责通知所有的broker更新元数据信息3.当使用kafka-topics.sh脚本为某个topic增加分区数量时，同样还是由控制器负责让新分区被其他节点感知到

Controller选举机制

Kafka核心总控制器Controller

controller感知到分区Leader所在的broker挂了(controller监听了很多zk节点可以感知到broker存活)controller会从ISR(In-sync replica)列表(参数unclean.leader.election.enable=false)里挑第一个broker作为leader（第一个broker最先放进ISR列表，可能时同步数据最多的副本）如果参数unclean.leader.election.enable为true，代表ISR列表里所有副本都挂了的时候可以在ISR列表以外的副本中选leader这种设置，可以提高可用性，但是选出的新leader有肯恩那个数据少很多，副本进入ISR列表有两个条件:1.副本节点不能那个产生分区，必须能与zk保持会话以及跟leader副本网络连通2.副本能复制leader上的所有写操作，并且不能落后太多。(与leader副本同步之后的副本，是由replica.lag.time.max.ms配置决定的，超过这个时间都没有跟leader同步过的一次的副本会被移出ISR列表)

Partition副本选举Leader机制

消费者消费消息的offset记录机制

rebalance就是说如果消费组里的消费者数量有变化或者消费的分区数有变化，fafka会重新分配消费者消费分区的关系比如consumer group中某个消费者挂了，此时会自动把分配给它的分区交给其他的消费者，如果它又重启了，那么又会把一些分区重新交还给他注意:relablance只针对subscribe这种不指定分区消费的情况，如果通过assign这种消费方式指定了分区，kafka不会进行rebalance如下情况可能会触发消费者realance:1.消费组里的consumer增加或减少了2.动态给topic增加了分区3.消费者组订阅了更多的topicrebalance过程中，消费者无法从kafka消费消息，这对kafka的TPS会有影响，如果kafka集群内节点较多，比如数百个，那rebalance可能会耗时极多，所以尽量避免在系统高峰期的rebalance发生

消费者的rebalance机制

1.写入方式producer采用push模式将消息发布到broker，每条消息都被append到partition中，属于顺序写磁盘(顺序写磁盘效率比随机写内存要高，保障kafka吞吐率)2.消息路由producer发送消息到broker时，会根据分区算法选择将其存储到哪一个partition，路由机制为:2.1 指定了partition，则直接使用2.2 未指定partition但指定key，通过对key的value进行hash选出一个partition2.3 partition和key都未指定，使用轮询选出一个partition3.写入流程3.1 producer先从zookeeper的/brokers/.../state/节点找到该partition的leader3.2 producer消息发送给该leader3.3 leader将消息写入本地log3.4 followers从leader pull消息，写入本地log后向leader发送ACK3.5 leader收到所有的ISR中的replica的ACK后，增加HW(high watermark，最后commit的offset)并向producer发送ACK

producer发布消息机制

Kafka的设计是一个partition在同一时刻只能被一个消费者组中的一个消费者消费，用来保证partion的顺序性，如果一个partition让两个消费者同时消费，则无法保证一个parition里面的消息消费是有序的，反观RocketMQ中的MessageQueue可以被多个消费者消费，因为RocketMQ保证顺序消费的机制是通过将一类key的消息发送到同一个队列当中来保证有序的

consumer消费消息机制

由此可见，Kafka的复制机制既不是完全的同步复制，也不是单纯的异步复制。事实上，同步复制要求所有能工作的follower都复制完，这条消息才会被commit，这种方式极大的影响了吞吐率。而异步复制的方式下，follower异步的从leader复制数据，数据只要被leader写入log就被认为已经commit，这种情况下如果follower都还没有复制完，落后于leader时，突然leader宕机，则会丢失数据而Kafka的这种使用ISR的方式则很好的均衡了确保数据不丢失以及吞吐率，还可以设置消息发送端对于发出消息持久化机制参数acks的设置

模型图解释

HW与LEO机制

部分消息的offset索引文件，kafka每次往分区发4K(可配置)消息就会记录一条当前消息的offset到index中，如果要定位消息的offset会在这个文件里快速定位，再去log文件里找具体消息00000000000000000000.index

index文件

消息存储文件，主要存offset和消息体00000000000000000000.log

log文件

消息的发送时间索引文件，kafka每次往分区发4K（可配置）消息的发送时间戳与对应的offset到timeindex文件，如果需要按照时间来定位消息的offset，会先在这个文件里找00000000000000000000.log

timeindex文件

文件类型

KafkaBroker有一个参数，log.setgment.bytes，限定了每个日志段文件的大小，最大就是1GB.一个日志段文件满了，就自动开一个新的日志段文件来写入，避免单个文件过大，影响文件的读写性能，这个过程叫做log rolling，正在被写入的那个日志段文件，叫做active log segment

日志分段存储

设计机制

普通消息

顺序消息

延时消息

支持的消息类型

一个ConsumerGroup中，只能有一个消费者消费消息

集群消费

一个ConsumerGroup中，每个消费者都可以消费到消息

广播消费

消费模式

亿级流量模型设计

JVM参数设置

线上规划

1.commit log文件会受到所在机器的文件系统大小的限制，分区之后可以将不同的分区放在不同的机器上，相当于对数据做了分布式存储，理论上一个topic可以处理任意数量的数据2.提高并行度

为什么要对topic下数据进行分区存储?

如何在多个partition中保证顺序消费?

4个过程都有可能造成消息丢失

如果消费这边配置的是自动提交，万一消费到数据还没处理完，就自动提交offset了，但是此时consumer直接宕机了未处理完的数据丢失了，下次也消费不到了

消息丢失

发送消息如果配置了重试机制，比如网络抖动事件过长导致发送端发送超时，实际broker可能已经接收到消息，但发送方会重新发送消息

如果消费这边配置的是自动提交，刚拉取了一批数据处理了一部分，但还没来得及提交，服务挂了，下次重启又会拉取相同的一批数据重复处理一般消费端都是要做消息幂等处理的

消费重复

如果发送端配置了重试机制，Kafka不会等之前那条消息完全成功了才去发送下一条消息，这样就可能出现发送了1，2，3条2消息，第一条超时了，后面两条发送成功，再重试发送第一条消息，这时消息在broker端的顺序就是2，3，1了所以，是否一定要配置重试要根据业务情况而定。也可以用同步发送的模式取发消息，当然acks不能设置为0，这样也能保证消息从发送端到消费端全链路有序kafka保证全链路消息顺序消费，需要从发送端开始，将所有有序消息发送到同一个分区，然后用一个消费者去消费，但是这种性能比较低，可以在消费者端接收到消息后将需要保证顺序消费的几条消息发到内存队列(可以多搞几个)，一个内存队列开启一个线程顺序消费处理

消息乱序

消息积压

延时队列存储的对象是延时消息，所谓的\"延时消息\"是指消息被发送以后，并不想让消费者立刻获取，而是等待特定的时间后，消费者才能获取这个消息进行消费，延时队列的使用场景有很多。比如:1.在订单系统中，一个用户下单之后通常有30分钟的时候进行支付，如果30分钟之内没有支付成功，那么这个订单将进行异常处理，这时就可以使用延时队列来处理这些订单了2.订单完成1小时后通知用户进行评价

实现思路:发送延时消息时先把消息按照不同的延迟时间段发送到指定的队列中(topic_1s、topic_5s、topic_10s....2h)这个一般不能支持任意时间段的延时)，然后通过定时器进行轮询这些topic，查看消息是否到期，如果到期就把这个消息发送到具体业务处理的topic中，队列中消息越靠前的到期时间越早，具体来说就是定时器在一次消费过程中，对消息的发送时间做判断，看下是否延迟到对应时间了，如果到了就转发，如果还没到这一次定时任务就可以提前结束了

延时队列

如果某段时间对已消费消息计算的结果觉得有问题，可能是由于程序bug导致的计算错误，当程序buf修复后，这时可能需要对之前已消费的消息重新消费，可以指定从多久之前的消息回溯消费，这种可以用consumer的offsetForTimes、seek等方法指定从某个offset偏移量的消息开始消费

消息回溯

可以用kafka压测工具自己测试分区数不同，各种情况下的吞吐量#往test里发送一百万消息，每条设置1kb# throughout用来进行限流控制，当设定的值小于0时不限流，当设定的值大于0时，当发送的吞吐量大于该值时就会被阻塞一段时间bin/kafka-producer-perf-test.sh --topic test --num-records 1000000 --record-size 1024 --throughout -1--producer-props bootstrap.servers=192.168.65.60:9092 acks=1

分区数越多吞吐量越高吗

kafka生产者的幂等性:因为发送端充值导致的消息重复发送问题，kafka的幂等性可以保证重复发送的消息指接收一次，只需要在生产者加上参数props.put(\"enable.idempotence\

消息传递保障

Kafka的事务不同于RocketMQ，RocketMQ是保障本地事务(比如数据库)与MQ消息发送的事务一致性，Kafka的事务主要是保障一次发送多条消息的事务一致性(要么同时成功要么同时失败)一般在Kafka的流式计算场景用得多一点，比如，kafka需要对于给topic里的消息做不同的流式计算处理处理完分别发到不同的topic里，这些topic分别被不同的下游系统消费(比如HBase、Redis、ES等)这种我们肯定希望系统发送到多个topic的数据保持事务一致性

Kafka的事务

磁盘顺序读写

kafka的消费者底层通过操作系统的sendfile来实现零拷贝Kafka的生产者则是通过操作系统的mmap来实现零拷贝1.减少了两次内核与用户空间的数据拷贝2.减少了内核与用户空间上下文切换

网上很多人说sendfile是直接从内核读取缓冲区拷贝到网卡接口里面，也有人说拷贝到socket缓冲区当中，我们通过```mand systemcalls sendfile```会发现，在Linux 内核2.6.33之前是拷贝到socket缓冲区当中，之后的版本是直接拷贝到了网卡接口

子主题

数据传输的零拷贝

读写数据的批量batch处理以及压缩传输

Kafka高性能的原因

线上问题及优化

Kafka