厦门网站建设开发公司,立方米网站建设,电影介绍网页设计代码,怎么建单位的网站首先Kafka是一个分布式消息队列中间件#xff0c;Apache顶级项目#xff0c;https://kafka.apache.org/ 高性能、持久化、多副本备份、横向扩展。生产者Producer往队列里发送消息#xff0c;消费者Consumer从队列里消费消息#xff0c;然后进行业务逻辑。应用场景主要有Apache顶级项目https://kafka.apache.org/ 高性能、持久化、多副本备份、横向扩展。生产者Producer往队列里发送消息消费者Consumer从队列里消费消息然后进行业务逻辑。应用场景主要有解耦、削峰缓冲、异步处理、排队、分布式事务控制等等。Kafka对外使用Topic(主题)的概念生产者往Topic里写消息消费者从Topic中消费读消息。为了实现水平扩展一个Topic实际是由多个Partition分区组成的遇到瓶颈时可以通过增加Partition的数量来进行横向扩容。单个Parition内是保证消息有序。持久化时每收到一条消息Kafka就是在对应的日志文件Append写所以性能非常高。Kafka Data Flow 消息流转图上图中消息生产者Producers往Brokers里面的指定Topic中写消息消息消费者Consumers从Brokers里面消费指定Topic的消息然后进行业务处理。在实际的部署架构中Broker、Topic、Partition这些元数据保存在ZooKeeper中Kafka的监控、消息路由分区由ZooKeeper控制。0.8版本的OffSet也由ZooKeeper控制。一、消息生产/发送过程Kafka创建Message、发送时要指定对应的Topic和Value消息体Key分区键和Partition分区是可选参数。 调用Producer的Send()方法后消息先进行序列化消息序列化器可自定义实现例如Protobuf然后按照Topic和Partition临时放到内存中指定的发送队列中。达到阈值后然后批量发送。发送时当Partition没设置时如果设置了Key-分区键例如单据类型按照Key进行Hash取模保证相同的Key发送到指定的分区Partition。如果未设置分区键Key使用Round-Robin轮询随机选分区Partition。二、分区Partition的高可用和选举机制分区有副本的概念保证消息不丢失。当存在多副本的情况下会尽量把多个副本分配到不同的broker上。Kafka会为Partition选出一个Leader Broker通过ZooKeeper之后所有该Partition的请求实际操作的都是Leader然后再同步到其他的Follower。当一个Kafka Broker宕机后所有Leader在该Broker上的Partition都会重新选举在剩余的Follower中选出一个Leader继续提供服务。正如上面所讲Kafka使用ZooKeeper在多个Broker中选出一个Controller用于Partition分配和Leader选举。以下是Partition的分配机制将所有Broker假设共n个Broker和待分配的Partition排序将第i个Partition分配到第i mod n个Broker上 这个就是leader将第i个Partition的第j个Replica分配到第(i j) mode n个Broker上Controller会在ZooKeeper的/brokers/ids节点上注册Watch一旦有broker宕机它就能知道。当Broker宕机后Controller就会给受到影响的Partition选出新Leader。Controller从ZooKeeper的/brokers/topics/[topic]/partitions/[partition]/state中读取对应Partition的ISRin-sync replica已同步的副本列表选一个出来做Leader。选出Leader后更新ZooKeeper的存储然后发送LeaderAndISRRequest给受影响的Broker进行通知。如果ISR列表是空那么会根据配置随便选一个replica做Leader或者干脆这个partition就是宕机了。如果ISR列表的有机器但是也宕机了那么还可以等ISR的机器活过来。多副本同步服务端这边的处理是Follower从Leader批量拉取数据来同步。但是具体的可靠性是由生产者来决定的。生产者生产消息的时候通过request.required.acks参数来设置数据的可靠性。 在acks-1的时候如果ISR少于min.insync.replicas指定的数目那么就会返回不可用。 这里ISR列表中的机器是会变化的根据配置replica.lag.time.max.ms多久没同步就会从ISR列表中剔除。以前还有根据落后多少条消息就踢出ISR在1.0版本后就去掉了因为这个值很难取在高峰的时候很容易出现节点不断的进出ISR列表。 从ISA中选出leader后follower会从把自己日志中上一个高水位后面的记录去掉然后去和leader拿新的数据。因为新的leader选出来后follower上面的数据可能比新leader多所以要截取。这里高水位的意思对于partition和leader就是所有ISR中都有的最新一条记录。消费者最多只能读到高水位 从leader的角度来说高水位的更新会延迟一轮例如写入了一条新消息ISR中的broker都fetch到了但是ISR中的broker只有在下一轮的fetch中才能告诉leader。 也正是由于这个高水位延迟一轮在一些情况下kafka会出现丢数据和主备数据不一致的情况0.11开始使用leader epoch来代替高水位。三、消息消费过程订阅topic是以一个消费组来订阅的一个消费组里面可以有多个消费者。同一个消费组中的两个消费者不会同时消费一个partition。换句话来说就是一个partition只能被消费组里的一个消费者消费但是可以同时被多个消费组消费。因此如果消费组内的消费者如果比partition多的话那么就会有个别消费者一直空闲。 消息Offset偏移量(消息的顺序号)管理 一个消费组消费partition需要保存offset记录消费到哪以前保存在zk中由于zk的写性能不好以前的解决方法都是consumer每隔一分钟上报一次。 ZooKeeper的性能严重影响了消费的速度而且很容易出现重复消费。 在0.10版本后Kafka把这个offset的保存从ZooKeeper总剥离保存在一个名叫__consumeroffsets topic的Topic中。 消息的key由groupid、topic、partition组成value是偏移量offset。topic配置的清理策略是compact。总是保留最新的key其余删掉。 一般情况下每个key的offset都是缓存在内存中查询的时候不用遍历partition如果没有缓存第一次就会遍历partition建立缓存然后查询返回。 Partitin的Rebalance 生产过程中broker要分配partition消费过程这里也要分配partition给消费者。类似broker中选了一个controller出来消费也要从broker中选一个coordinator用于分配partition。 coordinator的选举过程看offset保存在那个partition该partition leader所在的broker就是被选定的coordinator 交互流程 consumer启动、或者coordinator宕机了consumer会任意请求一个broker发送ConsumerMetadataRequest请求broker会按照上面说的方法选出这个consumer对应coordinator的地址。consumer 发送heartbeat请求给coordinator返回IllegalGeneration的话就说明consumer的信息是旧的了需要重新加入进来进行reblance。返回成功那么consumer就从上次分配的partition中继续执行。 Rebalanceconsumer给coordinator发送JoinGroupRequest请求。这时其他consumer发heartbeat请求过来时coordinator会告诉他们要reblance了。其他consumer发送JoinGroupRequest请求。所有记录在册的consumer都发了JoinGroupRequest请求之后coordinator就会在这里consumer中随便选一个leader。然后回JoinGroupRespone这会告诉consumer你是follower还是leader对于leader还会把follower的信息带给它让它根据这些信息去分配partitionconsumer向coordinator发送SyncGroupRequest其中leader的SyncGroupRequest会包含分配的情况。coordinator回包把分配的情况告诉consumer包括leader。 当partition或者消费者的数量发生变化时都得进行reblance。 列举一下会reblance的情况增加Partition增加消费者消费者主动关闭消费者宕机coordinator宕机四、消息投递语义kafka支持3种消息投递语义At most once最多一次消息可能会丢失但不会重复At least once最少一次消息不会丢失可能会重复Exactly once只且一次消息不丢失不重复只且消费一次0.11中实现仅限于下游也是kafkaAt least once业务中使用比较多先获取数据再进行业务处理业务处理成功后commit offset。生产者生产消息异常消息是否成功写入不确定重做可能写入重复的消息消费者处理消息业务处理成功后更新offset失败消费者重启的话会重复消费At most once先获取数据再commit offset最后进行业务处理。生产者生产消息异常不管生产下一个消息消息就丢了消费者处理消息先更新offset再做业务处理做业务处理失败消费者重启消息就丢了。Exactly once首先要保证消息不丢再去保证不重复。所以盯着At least once的原因来搞。 生产者重做导致重复写入消息----生产保证幂等性消费者重复消费---消灭重复消费或者业务接口保证幂等性重复消费也没问题业务处理的幂等性非常重要。Kafka控制不了需要业务来实现。比如所判断消息是否已经处理。解决重复消费有两个方法下游系统保证幂等性重复消费也不会导致多条记录。把commit offset和业务处理绑定成一个事务。生产的幂等性为每个producer分配一个pid作为该producer的唯一标识。producer会为每一个topic,partition维护一个单调递增的seq。类似的broker也会为每个pid,topic,partition记录下最新的seq。当req_seq broker_seq1时broker才会接受该消息。因为消息的seq比broker的seq大超过时说明中间有数据还没写入即乱序了。消息的seq不比broker的seq小那么说明该消息已被保存。 事务性和原子性 场景是这样的先从多个源topic中获取数据。做业务处理写到下游的多个目的topic。更新多个源topic的offset。 其中第2、3点作为一个事务要么全成功要么全失败。这里得益与offset实际上是用特殊的topic去保存这两点都归一为写多个topic的事务性处理。 引入tidtransaction id和pid不同这个id是应用程序提供的用于标识事务和producer是谁并没关系。就是任何producer都可以使用这个tid去做事务这样进行到一半就死掉的事务可以由另一个producer去恢复。 同时为了记录事务的状态类似对offset的处理引入transaction coordinator用于记录transaction log。在集群中会有多个transaction coordinator每个tid对应唯一一个transaction coordinator。 注transaction log删除策略是compact已完成的事务会标记成nullcompact后不保留。 启动事务时先标记开启事务写入数据全部成功就在transaction log中记录为prepare commit状态否则写入prepare abort的状态。之后再去给每个相关的partition写入一条markercommit或者abort消息标记这个事务的message可以被读取或已经废弃。成功后 在transaction log记录下commit/abort状态至此事务结束。 整体的数据流是这样的 首先使用tid请求任意一个broker代码中写的是负载最小的broker找到对应的transaction coordinator。请求transaction coordinator获取到对应的pid和pid对应的epoch这个epoch用于防止僵死进程复活导致消息错乱当消息的epoch比当前维护的epoch小时拒绝掉。tid和pid有一一对应的关系这样对于同一个tid会返回相同的pid。client先请求transaction coordinator记录topic,partition的事务状态初始状态是BEGIN如果是该事务中第一个到达的topic,partition同时会对事务进行计时client输出数据到相关的partition中client再请求transaction coordinator记录offset的topic,partition事务状态client发送offset commit到对应offset partition。client发送commit请求transaction coordinator记录prepare commit/abort然后发送marker给相关的partition。全部成功后记录commit/abort的状态最后这个记录不需要等待其他replica的ack因为prepare不丢就能保证最终的正确性了。 这里prepare的状态主要是用于事务恢复例如给相关的partition发送控制消息没发完就宕机了备机起来后producer发送请求获取pid时会把未完成的事务接着完成。 当partition中写入commit的marker后相关的消息就可被读取。所以kafka事务在prepare commit到commit这个时间段内消息是逐渐可见的而不是同一时刻可见。 消息消费事务 消费时partition中会存在一些消息处于未commit状态即业务方应该看不到的消息需要过滤这些消息不让业务看到kafka选择在消费者进程中进行过来而不是在broker中过滤主要考虑的还是性能。kafka高性能的一个关键点是zero copy如果需要在broker中过 滤那么势必需要读取消息内容到内存就会失去zero copy的特性。 五、 Kafka文件组织 kafka的数据实际上是以文件的形式存储在文件系统的。topic下有partitionpartition下有segmentsegment是实际的一个个文件topic和partition都是抽象概念。 在目录/${topicName}-{$partitionid}/下存储着实际的log文件即segment还有对应的索引文件。 每个segment文件大小相等文件名以这个segment中最小的offset命名文件扩展名是.logsegment对应的索引的文件名字一样扩展名是.index。有两个index文件一个是offset index用于按offset去查message一个是time index用于按照时间去查其实这里可以优化合到一起下面只说offset index。总体的组织是这样的 为了减少索引文件的大小降低空间使用方便直接加载进内存中这里的索引使用稀疏矩阵不会每一个message都记录下具体位置而是每隔一定的字节数再建立一条索引。 索引包含两部分分别是baseOffset还有position。 baseOffset意思是这条索引对应segment文件中的第几条message。这样做方便使用数值压缩算法来节省空间。例如kafka使用的是varint。 position在segment中的绝对位置。 查找offset对应的记录时会先用二分法找出对应的offset在哪个segment中然后使用索引在定位出offset在segment中的大概位置再遍历查找message。六、Kafka常用配置项 Broker配置 Topic配置 参考链接123archu 链接https://www.jianshu.com/p/d3e963ff8b70 原文地址https://www.cnblogs.com/tianqing/p/10808717.html.NET社区新闻深度好文欢迎访问公众号文章汇总 http://www.csharpkit.com
