一个kafka集群中包含一个或多个Producer、一个或多个broker、一个或多个ConsumerGrop以及一个Zookeeper集群。kafka通过Zookeeper管理kafka集群配置、leader副本的选举、生产者的负载均衡等。Producer使用push模式将消息发布到brokerConsumer使用pull模式从broker订阅并消费消息。专业术语kafkaCluster: kafka集群由一个或多个Broker节点组成。Broker: 一个Kafka集群包括一个或多个服务器一台服务器就是一个Broker节点。Broker用于保存Producer发送的消息。Producer:生产者用来发送指定的Topic的消息到Broker。生产者可以是代码还可以是命令行工具。本质上是一个进程或者线程。Consumer:消费者用来接收/消费Kafka集群中的消息。每个Consumer属于一个ConsumerGroup如果在创建消费者时没有指定Consumer,系统会默认分配一个ConsumerGroup消费者可以是代码还可以是命令行工具本质上就是一个进程/线程。ConsumerGroup:消费者组由一个或多个Consumer组成在同一个消费者组的消费者具有相同的group.id便于管理Consumer。Zookeeper:在Kafka集群中用来存储元数据如有Broker节点信息、分区的信息、分区与Broker的对应关系、生产者的负载均衡等等。Topic:主题主题用于区分业务比如订单主题业务购物车主题业务物流主题业务……方便对消息进行分类管理Partition:分区一个Topic的消息由一个或多个Partition存储。分区的作用是提高读写并行度/读写效率。Segment:分段,发送到kafka集群的消息会先存到内存中然后划分文件夹、划分文件存入磁盘中Replication:副本副本的作用是保证数据的安全性副本分为Leader(主副本)和Follower(从副本)Leader只有一个Follower可以有多个但是副本数一般都为1-3个副本数过多会占用大量的存储空间。注意读写都只能从Leader进行Follower在Leader宕机后自动选举出新的Leader。扩展 为什么读写都只能从Leader进行答保证数据的一致性只在Leader中进行写入数据Follow同步Leader中的数据在写过程中避免了多个副本中存储的数据不同的问题。Leader 和 Follow之间同步数据存在延时所以读操作也需要在Leader中进行。11. ISR: 表示目前Alive(活着的)且与Leader能够 “Catch-up”跟得上的Replicas(Follower)集合。12. Record:记录就是发送到Kafka集群的消息。一条消息就是一条记录。13. offset: 偏移量用于记录消息的序号各个分区的偏移量都是从0开始。备注Kafka中有分区和分段的概念分区就是分文件夹分段就是分文件。这个思想在Hive中也有:Hive中的分区就是分文件夹,Hive中的分桶就是分文件。跟RocketMQ的区别1元素据管理RocketMQ 有自己的管理节点 Nameserver, Kafka中是依赖zookeeper做元素据的管理。2消息存储RocketMQ 所有Topic数据同时只会写一个文件。Kafka每个Topic会对应多个Partition, 每个Partition又对应多个Segment 每个Segment就会写一个文件。3, 消息读写RocketMQ的broker 主节点支持读写从节点只读。 Kafka的Broker主节点读写从节只负责备份不提供服务。4消费模式消费模式Kafka 仅支持拉 (Pull)模式基于消费者组Consumer Group消费。RocketMQ 同时支持拉 (Pull)和推 (Push)两种模式支持集群和广播消费。5数据副本存放模式不一样同一个Topic的不同分区都会分散到不同的broker上。RocketMQ集群的broker区分主从节点主节点只放主队列从节点只放副本队列。Kafka集群的broker 不区分主从节点所有节点地位平等互相备份副本分区。6消息批量发送机制kafka 会自动攒消息到一个批次让后一次性发送RocketMQ需要手动编码将多条消息打包成一个消息包然后发送。7RocketMQ功能更丰富RocketMQ有事务消息定时消息 消息重试机制死信队列等Kafka都没有8 消息确认机制对比维度Apache KafkaApache RocketMQ生产者确认机制ACKs 级别配置通过acks参数控制。-acks0不等待确认性能最高可靠性最低。-acks1Leader 写入成功即确认性能与可靠性折中。-acksall-1所有 ISR 副本写入成功才确认可靠性最高。发送方式决定通过不同的发送模式提供不同保障。-同步发送阻塞等待 Broker 返回结果可靠性高。-异步发送通过回调获取结果不阻塞。-单向发送不关心结果性能最高可靠性最低。消费者确认机制位移Offset提交消费者处理完消息后提交消费进度Offset。-自动提交按固定间隔自动提交可能导致重复消费。-手动提交业务处理完成后显式调用commitSync()或commitAsync()提交。状态返回消费者处理完消息后向 Broker 返回处理状态。-CONSUME_SUCCESS确认成功。-RECONSUME_LATER消费失败需要重试。9 刷盘策略刷盘策略Kafka 主要依赖操作系统异步刷盘消息先写入 PageCache由操作系统根据自身策略或log.flush.*参数决定何时调用fsync落盘。默认情况下不强制主动刷盘以最大化吞吐。RocketMQ 提供两种明确策略1.同步刷盘 (SYNC_FLUSH)消息写入 PageCache 后立即强制落盘成功后返回可靠性极高但性能有损耗。2.异步刷盘 (ASYNC_FLUSH)消息写入 PageCache 后立即返回由后台线程定时如每500ms批量刷盘性能高但宕机时可能丢失少量消息。架构详解1.生产者生产者采用push(推模式)向集群发送消息并且消息是被顺序写磁盘追加到分区中提高了kafka的写效率吞吐量备注顺序写效率随机写效率生产者只需要连接上任意一个活着的Broker就可以连接上Kafka集群每个Broker都缓存了一份完整的集群元数据。这意味着无论客户端连接到哪个Broker这个Broker都能提供客户端所需的全部集群信息。生产者发送消息时可以指定Topic、分区编号、key、value分区编号和key都可以决定消息或者说是记录进入到哪个分区具体规则如下没有key,默认轮询方式写入到分区有key,没有分区编号,使用key的hash % 分区数得到分区编号有key,有分区编号,直接使用指定的分区编号也可以使用自定义分区策略,可以参考DefaultPartitioner实现Partitioner接口即可分区的作用提高读写效率/并行度方便集群扩展,业务扩张,数据增加的时候,可以增加机器,并增加分区数,以提升Kafka处理能力分区的目的是为了提高并行度,数据的安全由副本保证,且副本是以分区来备份的!所以就有了:partition0的leader副本,partition0的follow副本! 注意:只能从leader读写,follow只负责备份消息是局部有序分区内有序生产者发送消息到Kafka的各个分区中根据消息发送的分区策略不能保证发送的分区是有序的但是在分区内按照offset的顺序追加写的。每个主题的每个分区中offset都是从0开始。消息确认机制acks0意思就是KafkaProducer客户端只要把消息发送出去不管那条数据有没有在Partition Leader上落到磁盘都不管他了直接就认为这个消息发送成功了acks1只要Partition Leader接收到消息就认为成功了不管他其他的Follower有没有同步过去这条消息了。acksall/-1意思就是说Partition Leader接收到消息之后还必须要求ISR列表里跟Leader保持同步的那些Follower都要把消息同步过去才能认为这条消息是写入成功了。all即所有副本都同步到数据时send方法才返回, 以此来完全判断数据是否发送成功, 理论上来讲数据不会丢失2.Broker节点分区体现在分文件夹分段体现在分文件数据发送到Kafka集群最终会存储在分区下的分段中也就是partition下的segment文件中而kafka是一个消息系统并不是一个存储系统所以这些消息/数据/记录是有生命周期默认配置如下所示1基于时间log.retention.hours168 # 7天2基于大小log.retention.bytes1073741824 #单个segment达到1个G满足上面任意一个就会被删除!如何根据offset从分段文件中找到需要读取的数据由于生产者生产的消息会不断的追加到log文件的末尾为防止log文件过大导致数据定位效率低下kafka采取了分片和索引机制将每个partition分为多个segment每个segment对应两个文件 .index文件和 .log文件这些文件位于一个文件夹下该文件夹的命名规则为topic名称分区序号。例如first这个topic有三个分区则其对应的文件夹为first-0,first-1,first-2.其中文件内容如下00000000000000000000.index //每个segment都有一个.index文件一个.log文件 00000000000000000000.log 00000000000000170410.index 00000000000000170410.log 00000000000000239430.index 00000000000000239430.logindex文件和log文件以当前segment的第一条消息的offset命名。什么意思呢以上面的文件内容举例00000000000000000000.index 说明第一条消息offset是0也就是最开始的位置00000000000000170410.index 这里偏移量是170410说明是170410这个位置开始。index和log文件示意图第1步确定所在的Log Segment第2步在Index文件中进行二分查找找到目标Segment后接下来打开对应的.index索引文件。索引结构.index文件的内容是偏移量和物理位置的键值对。这些条目是根据Offset有序存储的。偏移量Relative Offset 注意这里存储的通常是相对于该Segment基础Offset的相对偏移量以节省空间4字节但理解原理时可以认为是绝对Offset。物理位置Position 对应消息在.log文件中的字节偏移量。查找过程 对.index文件中的索引条目进行二分查找寻找一个特定的索引条目该条目的Offset ≤O_target并且是满足这个条件的最大Offset。为什么不是直接找相等的因为索引文件是稀疏索引不会为每一条消息都建索引而是每隔一定数量的消息由log.index.interval.bytes配置才建一条索引。所以很可能找不到offset的直接对应项。第3步从Log文件中的指定位置开始顺序扫描上一步的二分查找会给我们一个关键的物理位置信息假设为P_index。定位 Kafka会打开对应的.log数据文件并从P_index这个字节位置开始顺序向下扫描。扫描 因为消息是顺序存储的并且每条消息都有固定的格式包含消息长度、Offset等信息所以Kafka可以高效地解析每条消息并比较其Offset。命中 当扫描到某条消息的Offset等于O_target时查找成功返回该消息。异常 如果扫描直到Segment末尾都没有找到该Offset可能是因为该Offset不存在例如超出了最大Offset则会返回错误。举例说明假设有一个Segment文件.log文件00000000000000000000.log.index文件00000000000000000000.index.index文件内容可能如下简化表示Offset (相对)Position00// 第一条消息在.log文件的0字节处500102400// Offset500的消息在.log文件的102400字节处1000204800// 以此类推...现在我们要查找Offset 755的消息。确定Segment 根据文件名该Segment包含Offset从0开始的消息所以755在其中。查询.index 在.index文件中二分查找找到小于等于755的最大索引条目是Offset500, Position102400。扫描.log 从.log文件的102400字节处开始顺序读取消息。Kafka会读出一条条消息并判断它们的Offset读到Offset500的消息然后Offset501, 502, 503 ... 753, 754, 755找到 当读到Offset755的消息时停止扫描将该消息返回给客户端。性能为何高效磁盘顺序访问 尽管有二分查找但最终是从索引定位到的位置开始顺序读取磁盘这种操作模式的速度非常快甚至可以媲美内存随机读取。稀疏索引 索引文件很小可以完全映射到内存中二分查找的过程是在内存中进行的速度极快。分片机制 通过Segment将大文件分割成小文件使得查找过程中的二分查找效率更高不需要处理巨大的单个文件。3.消费者细节1.消费者从Kafka消费消息,使用的pull拉取模式注意:Kafka为什么生产者是push推模式,消费者是pull拉模式,因为这样Kafka的压力较小,性能较高!类似与日常生活中的快递柜!2.消费者只需要连接上任意一个活着的Broker就可以连接上整个Kafka集群 每个Broker都缓存了一份完整的集群元数据。这意味着无论客户端连接到哪个Broker这个Broker都能提供客户端所需的全部集群信息。3.消费者消费消息的时候可以指定从哪个offset开始消费,如果有记录则从记录的位置开始消费,如果没有记录,取决于auto.offset.reset,值为earliest表示从最开始的数据,latest表示从最新的数据,none报错4.offset可以自动提交也可以手动提交: 消费者提交Offset就是消费者向Kafka“汇报”它已经成功处理到了哪个位置的消息。//是否自动提交offset,true表示自动提交 props.put(enable.auto.commit, true); //自动提交偏移量时的时间间隔ms值 props.put(auto.commit.interval.ms, 1000);5.老版本offset提交到zk中,新版本提交到默认主题__consumer_offsets中6.消费者消费消息的时候可以指定订阅主题 也可以指定订阅主题和分区7.消费者可以指定消费者组名,方便对消费者进行管理为什么需要提交Offset主要为了解决两个问题消息丢失和消息重复。防止消息丢失At least once 语义的基础假设消费者拉取了一批消息Offset 0-99正在处理时突然崩溃了。如果它没有提交Offset比如提交到Offset 100当它重启或者由其他消费者实例接管它的工作时Kafka会认为这批消息从未被处理过。新的消费者会再次拉取Offset 0-99的消息。这样消息至少被处理了一次虽然可能重复但绝不会丢失。记录消费进度避免永远重头开始如果没有Offset提交机制每次消费者启动都只能从最新的latest或最早的earliest消息开始消费完全无法记录之前的进度。这对于需要处理历史数据的系统来说是灾难性的。提交的内容是什么消费者会向一个特殊的Kafka内部Topic名为__consumer_offsets发送一条消息。这条消息的内容大致是“消费者组my-group对于Topicmy-topic的分区0最后成功处理的消息Offset是123”Kafka会持久化这个信息。之后同属这个消费者组的任何消费者实例都可以来查询这个进度。提交的时机和方式非常重要提交的时机和策略直接决定了系统是“至少一次”、“至多一次”还是“精确一次”语义。方式描述优点缺点消息语义自动提交enable.auto.committrue消费者在后台周期性地auto.commit.interval.ms自动提交最后拉取的一批消息的Offset。使用简单无需手动管理。容易导致重复消费如果刚拉取完消息自动提交的时间还没到消费者崩溃了。重启后Offset还没提交会重新拉取那批已经处理过的消息。至少一次同步手动提交consumer.commitSync()在处理完一批消息后主动调用commitSync()方法。该方法会阻塞直到Broker确认提交成功。更安全确保处理完再提交避免自动提交的“未提交则重复”问题。性能较差因为会阻塞消费者线程等待Broker确认。至少一次异步手动提交consumer.commitAsync()在处理完一批消息后主动调用commitAsync()方法。该方法立即返回不会阻塞。性能好不会阻塞消费者线程。如果提交失败不会自动重试可能导致提交乱序后一次的提交先成功有覆盖Offset的风险。至少一次同步异步组合提交常见的可靠模式。在正常的循环中使用commitAsync()保证性能在关闭消费者或再均衡前使用commitSync()确保提交成功。兼顾了性能和可靠性。实现稍复杂。至少一次