Kafka

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

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

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

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

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

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

在Kafka集群中会有一个或者多个broker，其中有一个broker会被选举为控制器(Kafka Controller)，
它负责管理整个集群中所有分区和副本的状态

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

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列表)

每个consumer会定期将自己消费分区的offset提交给kafka内部topic:_consumer_offsets，
提交过去的时候，key时consumerGroupId+topic+分区号，value就是当前offset的值，
kafka会定期清理topic里的消息，最后就保留最新的那条数据，因为_consumer_offset可能会接收高并发的请求，
kafka默认给其分配50个分区(可以通过offsets.topic.num.paritions设置),这样可以通过加机器的方式抗大并发

rebalance就是说如果消费组里的消费者数量有变化或者消费的分区数有变化，kafka会重新分配消费者消费分区的关系
比如consumer group中某个消费者挂了，此时会自动把分配给它的分区交给其他的消费者，如果它又重启了，那么
又会把一些分区重新交还给他
注意:relablance只针对subscribe这种不指定分区消费的情况，如果通过assign这种消费方式指定了分区，kafka不会进行rebalance
如下情况可能会触发消费者realance:

• 1.消费组里的consumer增加或减少了
• 2.动态给topic增加了分区
• 3.消费者组订阅了更多的topic
• rebalance过程中，消费者无法从kafka消费消息，这对kafka的TPS会有影响，如果kafka集群内节点较多，比如数百个，
• 那rebalance可能会耗时极多，所以尽量避免在系统高峰期的rebalance发生

1.写入方式
producer采用push模式将消息发布到broker，每条消息都被append到partition中，属于顺序写磁盘(顺序写磁盘效率比随机写内存要高，保障kafka吞吐率)
2.消息路由
producer发送消息到broker时，会根据分区算法选择将其存储到哪一个partition，路由机制为:

• 2.1 指定了partition，则直接使用
• 2.2 未指定partition但指定key，通过对key的value进行hash选出一个partition
• 2.3 partition和key都未指定，使用轮询选出一个partition
  

3.写入流程

• 3.1 producer先从zookeeper的/brokers/.../state/节点找到该partition的leader
• 3.2 producer消息发送给该leader
• 3.3 leader将消息写入本地log
• 3.4 followers从leader pull消息，写入本地log后向leader发送ACK
• 3.5 leader收到所有的ISR中的replica的ACK后，增加HW(high watermark，最后commit的offset)并向producer发送ACK

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

HW俗称高水位，HighWatermark的缩写，取一个partition对应的ISR中最小的LEO(log-end-offse)作为HW，
consumer最多只能消费到HW所在的位置，另外每个replica都有HW,leader和follower各自负责更新自己的状态
对于leader新写入的消息，consumer不能立刻消费，leader会等待该消息被所有ISR中的replica同步后更新HW
此时消息才能被Consumer消费，这样就保证了如果leader所在的broker失效，该消息仍然可以从新选举的leader中获取
对于来自内部broker的读取请求，没有HW的限制

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

模型图解释

Kafka一个分区的消息数据对应存储在一个文件夹下，以topic名称+分区号命名，消息在分区内是分段(Segment)存储，
每个段的消息都存储在不一样的log文件里，这种特性方便old segment file快速被删除，kafka规定了一个段位的log文件
最大为1G,做这个限制的目的是为了方便把log文件加载到内存去操作

文件类型

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

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

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

Kafka是由Scala语言开发，运行在JVm上，需要对JVM参数合理设置
修改/bin/kafka-start-server.sh中的jvm设置，假设机器是32G内存，可进行如上设置:

这种大内存的情况一般都要用G1垃圾收集器,因为年轻代内存比较大，用G1可以设置GC最大停顿时间，不至于
一次minor gc就花费太长时间，当然,因为像Kafka、RocketMQ ES这些中间件，写数据到磁盘会用到操作系统
的Page Cache，所以JVM内存不宜分配过大，需要给操作系统的缓存预留出几个G

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

方案一:首先将需要保证顺序的消息收集起来，然后交给一个consumer去进行处理，然后内部维护一个线程池，让其中某一个线程去顺序执行这些消息eg:用户下单流程，支付成功消息 -> 库存消息
方案二:让多个消息构造一个特殊结构的顺序消息,当consumer收到时，在一个线程中依次进行消费

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

Producer

Consumer

Producer

Consumer

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

一个parition同一时刻在一个consumer group中只能有一个consumer实例在消费
,从而保证消费顺序。consumer group中的consumer数量不能比一个topic中的partion数量还要多，否则
多出来的consumer消费不到消息。
Kafka只在parition的范围内保证消息消费的局部顺序性，不能在同一个topic中的多个partition中保证总的消费性
如果有在总体上保证消费顺序的需求，那么我们可以通过将topic的partition数量设置为1，将consumer group中的
consumer instance数量也设置为1，但是这样会影响性能，所以kafka的顺序消费很少用

1.线上有时因为发送方发送消息速度过快，或者消费放处理消息过慢，可能会导致broker挤压大量未消费消息
此种情况如果挤压了上百万未消费消息需要紧急处理，可以修改消费端程序，让其将收到地消息快速转发到其他
topic(可以设置很多分区),然后再启动多个消费者同时消费新主题地不同分区

2.由于消息数据格式变动或者消费者程序有bug，导致消费者一直消费不成功，也可能导致broker积压大量未
消费消息.此种情况可以将这些消费不成功地消息转发到其他队列里去(类似死信队列),后面再慢慢分析死信队列
里地消息处理问题

延时队列存储的对象是延时消息，所谓的"延时消息"是指消息被发送以后，并不想让消费者立刻获取，而是等待
特定的时间后，消费者才能获取这个消息进行消费，延时队列的使用场景有很多。比如:
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

网络上很多资料都说分区数越多吞吐量很高，但从压测结果来看，分区数到达某个值，吞吐量反而开始下降
实际上很多事情都会有一个临界值，当超过这个临界值之后，很多原本符合既定逻辑的走向又会变得不同,
一般情况下跟集群机器数量相当就差不多了，比如分区对文件描述符的占用，以及缓冲区的竞争，
类似于Redis集群数量不能超过1000个，当超过1000个时，
整体的网络心跳将会特别长，心跳数据包会比较大,心跳包中会包含，比较容易产生网络分区
当然吞吐量的数值和走势还会和磁盘、文件系统、IO调度策略等因素有关
注意:如果分区数设置过大，比如设置10000，可能会设置不成功，后台可能会报错
"java.io.IOException: Too many open files"异常中最关键的信息是"Too many open files",
这是一种常见的Linux系统错误，通常意味着文件描述符不足，它一般发生在创建线程、创建socket、
打开文件这些场景下，在Linux系统的默认设置下，这个文件描述符的个数不是很多，通过ulimit -n命令可以查看
一般默认是1024，可以将该值调大比如 ulimit -n 65535

at most once(消费者最多收到一次消息, 0-1次) acks = 0可以实现
at least once(消费者至少收到一次消息，1-多次) acks = all可以实现
exactly once(消费者刚好收到一次消息): at least once加上消费者幂等可以实现，还可以用kafka生产者的幂等性
来实现

kafka生产者的幂等性:
因为发送端充值导致的消息重复发送问题，kafka的幂等性可以保证重复发送的消息指接收一次，只需要
在生产者加上参数props.put("enable.idempotence",true)即可，默认是false不开启
具体实现原理是，kafka每次发送消息会生成PID和Sequence Number，并将这两个属性一起发送给broker
broker会将PID和Sequence Number跟消息绑定一起存起来，下次如果生产者重发相同消息，broker会检查
PID和Sequence Number，如果相同不会再接收

PID:每个新的Producer在初始化的时候会被分配一个唯一的PID,这个PID对用户完全是透明的，生产者
如果重启则会生成新的PID,
Sequence Number:对于每个PID，该Producer发送到每个Partition的数据都有对应的序列号，这些
序列号是从0开始递增的

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

磁盘顺序读写

数据传输的零拷贝

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