商服网站模板,计算机网站开发国外参考文献,上海 企业,北京vi设计培训文章目录 概览MQ优点MQ缺点常见MQ对比JMS消息模型点对点模式发布订阅模式 kafka基础架构发布订阅工作流程生产者生产者文件存储生产者分区策略生产者数据可靠性保证生产者数据一致性保证生产者ack机制ExactlyOnce生产者发送消息流程 消费者消费者分区分配策略消费者消费数据问题… 文章目录 概览MQ优点MQ缺点常见MQ对比JMS消息模型点对点模式发布订阅模式 kafka基础架构发布订阅工作流程生产者生产者文件存储生产者分区策略生产者数据可靠性保证生产者数据一致性保证生产者ack机制ExactlyOnce生产者发送消息流程 消费者消费者分区分配策略消费者消费数据问题 kafka事务消息积压 概览
MQ 即 messagequeue 消息队列是分布式系统的重要组件主要解决异步消息应用解耦消峰等问题。从而实现高可用高性能可伸缩和最终一致性的架构。使用较多的MQ有activeMQrabbitMQkafkametaMQ。
MQ优点
异步消息处理可以将一些非核心流程如日志短信邮件等通过MQ的方式异步去处理。这样做的好处是缩短主流程的响应时间提升用户体验应用解耦合商品服务和订单服务之间。用户下单后订单服务会通知商品服务。不使用MQ的情况是订单服务调用商品服务的接口这样订单服务和商品服务之间是耦合的使用MQ订单服务完成持久化处理将消息写入MQ消息队列中返回用户订单下单成功商品服务来订阅这个下单的消息采用拉或推的方式获得下单信息商品服务根据商品下单信息进行商品库存信息修改这样当下单时商品服务不可用时也不影响正常下单这就完成了订单服务和商品服务之间的解耦流量消峰秒杀活动流量过大导致流量暴增最终可能导致应用挂掉。一般会在应用前端加入消息队列来控制活动人数假如消息队列超过最大数量应该直接抛弃用户请求或者跳转到错误页面。秒杀业务根据消息队列中的请求信息在做后续的业务处理。比如在抢购时可能一下子过来了10万个请求但MQ只接受前100个用户的请求超过100个不接收了。这样就成功限制了用户请求
MQ缺点
系统可用性降低: 系统引入的外部依赖越多越容易挂掉;系统复杂度提高: 加个 MQ 进来你怎么保证消息没有重复消费怎么处理消息丢失的情况怎么保证消息传递的顺序性头大头大问题一大堆痛苦不已;一致性问题: A 系统处理完了直接返回成功了人都以为你这个请求就成功了但是问题是要是 BCD 三个系统那里BD 两个系统写库成功了结果 C 系统写库失败了你这数据就不一致了;
常见MQ对比
特性ActiveMQRabbitMQRocketMQKafka单机吞吐量万级比 RocketMQ、Kafka 低一个数量级同 ActiveMQ10 万级支撑高吞吐10 万级高吞吐一般配合大数据类的系统来进行实时数据计算、日志采集等场景topic 数量对吞吐量的影响topic 可以达到几百/几千的级别吞吐量会有较小幅度的下降这是 RocketMQ 的一大优势在同等机器下可以支撑大量的 topictopic 从几十到几百个时候吞吐量会大幅度下降在同等机器下Kafka 尽量保证 topic 数量不要过多如果要支撑大规模的 topic需要增加更多的机器资源时效性ms 级微秒级这是 RabbitMQ 的一大特点延迟最低ms 级延迟在 ms 级以内可用性高基于主从架构实现高可用同 ActiveMQ非常高分布式架构非常高分布式一个数据多个副本少数机器宕机不会丢失数据不会导致不可用消息可靠性有较低的概率丢失数据基本不丢经过参数优化配置可以做到 0 丢失同 RocketMQ功能支持MQ 领域的功能极其完备基于 erlang 开发并发能力很强性能极好延时很低MQ 功能较为完善还是分布式的扩展性好功能较为简单主要支持简单的 MQ 功能在大数据领域的实时计算以及日志采集被大规模使用
一般的业务系统要引入 MQ最早大家都用 ActiveMQ但是现在确实大家用的不多了没经过大规模吞吐量场景的验证社区也不是很活跃所以大家还是算了吧我个人不推荐用这个了。
后来大家开始用 RabbitMQ但是确实 erlang 语言阻止了大量的 Java 工程师去深入研究和掌控它对公司而言几乎处于不可控的状态但是确实人家是开源的比较稳定的支持活跃度也高。
不过现在确实越来越多的公司会去用 RocketMQ确实很不错毕竟是阿里出品但社区可能有突然黄掉的风险,目前 RocketMQ 已捐给 Apache但 GitHub 上的活跃度其实不算高,对自己公司技术实力有绝对自信的推荐用 RocketMQ否则回去老老实实用 RabbitMQ 吧人家有活跃的开源社区绝对不会黄。
所以中小型公司技术实力较为一般技术挑战不是特别高用 RabbitMQ 是不错的选择大型公司基础架构研发实力较强用 RocketMQ 是很好的选择。
如果是大数据领域的实时计算、日志采集等场景用 Kafka 是业内标准的绝对没问题社区活跃度很高绝对不会黄何况几乎是全世界这个领域的事实性规范。
JMS消息模型
JMS即JavaMessageServiceJava消息服务应用程序接口是一个Java平台中关于面向消息中间件的API用于在两个应用程序之间或分布式系统中发送消息进行异步通信。
JMS是基于JVM的消息代理规范ActiveMQ、HornetMQ等是JMS的实现。
Java消息服务是一个与具体平台无关的API绝大多数MOM提供商都对JMS提供支持。我们可以简单的理解两个应用程序之间需要进行通信我们使用一个JMS服务进行中间的转发通过JMS的使用我们可以解除两个程序之间的耦合。
点对点模式
消息发送者发送消息消息代理将其放入消息队列中消息接受者从队列中获取消息消息读取后被移除消息队列。每个消息都被发送到一个特定的队列接收者从队列中获取消息。队列保留着消息直到它们被消费或超时。 虽然可能有多个客户端在队列中侦听消息但只有一个可以读取到消息之后消息将不存在其他消费者将无法读取。也就是说消息队列只有唯一一个发送者和接受者但是并不能说只有一个接收者。
特点
每个消息只有一个消费者即消息一旦被消费消息就不在消息队列中发送者和接收者之间在时间上没有依赖性也就是说当发送者发送了消息之后不管接收者有没有正在运行它不会影响到消息被发送到队列接收者在成功接收消息之后需向队列应答成功
发布订阅模式
发布者将消息发送到主题Topic中多个订阅者订阅这个主题订阅者不断的去轮询监听消息队列中的消息那么就会在消息到达的同时接收消息。 特点
每个消息可以有多个消费者消费完消息之后消息不会清除发布者和订阅者之间有时间上的依赖性针对某个主题(Topic)的订阅者它必须创建一个订阅之后才能消费发布者的消息而且为了消费消息订阅者必须保持运行的状态。当然为了缓和这种严格的时间相关性JMS允许订阅者创建一个可持久化的订阅。这样即使订阅者没有运行它也能接收到发布者的消息
kafka
kafka是一个分布式的基于发布/订阅模式的消息队列主要应用于大数据实时处理领域。
基础架构 Producer消息生产者向kafka推送消息Consumer消息消费者向kafka中broker获取消息的客户端Topic主题可以理解为一个队列生产者和消费者面向的都是一个topicBroker经纪人一台kafka服务器就是一个broker。一个集群由多个broker组成。一个broker可以容纳多个topicPartition分区为了实现扩展性一个非常大的topic可以分布到多个服务器上一个主题可以分为多个分区每个分区是一个有序的队列ConsumerGroup消费者组由多个消费者组成。消费者组内每个消费者负责消费不同分区的数据一个分区只能由一个消费者组内的一个消费者消费消费者组之间互不影响。所有的消费者都属于某个消费者组即消费者组是逻辑上的一个订阅者Replica副本为保证集群中的某个节点发生故障时该节点上的分区数据不丢失且kafka仍然能够继续工作kafka提供了副本机制一个topic的每个分区都有若干个副本一个leader和若干个followerLeader每个分区多个副本的“主”生产者发送数据的对象以及消费者消费数据的对象都是leader。leader是针对topic的而不是broker的即不同的kafka服务区中会出现同一个leaderFollower每个分区多个副本中的“从”实时从leader中同步数据保持和leader数据的同步。leader发生故障时某个Follower会成为新的leaderControllerKafka 集群中有一个broker会被选举为Controller负责管理集群broker的上下线所有topic的分区副本分配和leader选举等工作Controller的管理工作都是依赖于Zookeeper的
消费者组的作用为了提高消费能力即提高并发。
解耦合是消息队列作用之一当消费者宕机后再次启动的时候会继续消费消息而不是从头消费消息。因为这个特性所以消费者会保存一些消费的进度信息被称为offset保存的位置在kafka0.9之前保存在zookpeer当中在此之后保存在kafka本地。即最终kafka会将消息保存在本地磁盘中默认保留168个小时即7天。
kafka中消息是以topic进行分类的producer生产消息consumer消费消息都是面向topic的。topic是逻辑上的概念而partition是物理上的概念每个partition对应于一个log文件该log文件中存储的就是producer生产的数据。
Producer生产的数据会被不断追加到该log文件末端且每条数据都有自己的offset。consumer组中的每个consumer都会实时记录自己消费到了哪个offset以便出错恢复时从上次的位置继续消费。
发布订阅工作流程 生产者定期向主题发送消息kafka代理将所有消息存储在为特定主题配置的分区中。它确保消息在分区之间平均共享。如果生产者发送了两个消息并且有两个分区kafka将在第一个分区中存储一个消息在第二个分区中存储第二个消息消费者订阅主题后kafka将向消费者提供该主题的当前偏移量并将偏移量保存在Zookeeper集合中消费者将定期向kafka请求新消息,一旦kafka从生产者那里收到消息它将把这些消息转发给消费者消费者将收到消息并进行处理消息处理后消费者将向kafka代理发送确认一旦kafka收到确认它将偏移量更改为新值并在Zookeeper中对其进行更新。由于在Zookeeper中保留了偏移量因此即使在服务器出现故障时使用者也可以正确读取下一条消息4、5、6步骤流程将重复进行直到消费者停止请求为止
消费者可以选择随时倒退或跳至所需的主题偏移量并阅读所有后续消息。
生产者
生产者文件存储
参考文章
https://www.jianshu.com/p/3e54a5a39683 由于生产者生产的消息会不断追加到log文件末尾为防止log文件过大导致数据定位效率低下kafka采取了分片和索引机制将每个partition分为多个segment。
每个segment对应两个文件“.index”文件和“.log”文件。这些文件位于一个文件夹下该文件夹的命名规则为topic名称分区序号。例如first这个topic有三个分区则其对应的文件夹为 first-0,first-1,first-2。其中每个segment中的日志数据文件大小均相等。 该日志数据文件的大小可以通过在kafka Broker的config/server.properties配置文件的中的“log.segment.bytes”进行设置默认为1G大小1073741824字节在顺序写入消息时如果超出该设定的阈值将会创建一组新的日志数据和索引文件。 “.index”文件存储大量的索引信息“.log”文件存储大量的数据索引文件中的元数据指向对应数据文件中message的物理偏移地址。其中文件的命名是以第一个数据的偏移量来命名的。
kafka如何通过index文件快速找到log文件中的数据
根据指定的偏移量使用二分法查询定位出该偏移量对应的消息所在的分段索引文件和日志数据文件。然后通过二分查找法继续查找出小于等于指定偏移量的最大偏移量同时也得出了对应的position即实际物理位置。
根据该物理位置在分段的日志数据文件中顺序扫描查找偏移量与指定偏移量相等的消息。由于index文件中的每条对应log文件中存储内容大小都相同所以想要找到指定的消息只需要用index文件中的该条的大小加上该条的偏移量即可得出log文件中指定消息的位置。 生产者分区策略
分区的原因
方便在集群中扩展每个Partition可以通过调整以适应它所在的机器而一个topic又可以有多个Partition组成因此整个集群就可以适应适合的数据了可以提高并发因为可以以Partition为单位读写了每个partition在不同的服务区上 在指明partition的情况下直接将指明的值直接作为partiton值在没有指明partition值但有key的情况下将key的hash 值与topic的partition数进行取余得到partition值既没有partition值又没有key值的情况下第一次调用时随机生成一个整数后面每次调用在这个整数上自增将这个值与topic可用的partition总数取余得到partition值也就是常说的round-robin算法
生产者数据可靠性保证
为保证 producer 发送的数据能可靠的发送到指定的 topictopic 的每个 partition 收到 producer 发送的数据后都需要向 producer 发送 ack(acknowledgement 确认收到)如果 producer 收到 ack就会进行下一轮的发送否则重新发送数据。 方案优点缺点半数以上完成同步就发 送 ack延迟低选举新的 leader 时容忍 n 台 节点的故障需要 2n1 个副本全部完成同步才发送ack选举新的 leader 时容忍 n 台 节点的故障需要 n1 个副 本延迟高 理解 2n1 半数以上完成同步才可以发ACK如果挂了n台有副本的服务器那么就需要有另外n台正常发送这样正常发送的刚好是总数挂的和没挂的的一半n挂的n正常的2n,因为是半数以上所以2n1.(所以总数2n1的时候最多只能容忍n台有故障) 即如果挂了n台有副本的服务器那么存在副本的服务器的总和为 2n1 kafka选择了第二种方案原因如下: 1.同样为了容忍 n 台节点的故障第一种方案需要 2n1 个副本而第二种方案只需要 n1 个副本而 kafka 的每个分区都有大量的数据第一种方案会造成大量数据的冗余 2.虽然第二种方案的网络延迟会比较高但网络延迟对 kafka 的影响较小
采用第二种方案之后设想以下情景 leader 收到数据所有 follower 都开始同步数据但有一个 follower因为某种故障迟迟不能与 leader 进行同步那 leader 就要一直等下去直到它完成同步才能发送 ack。这个问题怎么解决呢
Leader 维护了一个动态的 in-sync replica set 即ISR。
当和 leader 保持同步的 follower 集合。当 ISR 中的 follower 完成数据的同步之后就会给 leader 发送 ack。如果 follower长时间未向leader同步数据则该follower将被踢出ISR该时间阈值由replica.lag.time.max.ms参数设定。 Leader 发生故障之后就会从 ISR 中选举新的 leader。 kafka是通过消息条数差值(replica.lag.time.max.messages) 加 通信时间长短(同步时间replica.lag.time.max.ms) 两个条件来选副本进ISR在高版本中不再关注副本的消息条数最大条件。 为何会去掉消息条数差值参数
因为kafka一般是按batch批量发数据到leader, 如果批量条数12条replica.lag.time.max.messages参数设置是10条那么当一个批次消息发到kafka leader此时ISR中就要踢掉所有的follower很快follower同步完所有数据后follower又要被加入到ISR而且要加入到zookeeper中频繁操作所以去除掉该条件。
生产者数据一致性保证 LEOLog End offset每个副本的最后一个offset HWHigh Watermark高水位指的是消费者能见到的最大的 offset ISR 队列中最小的 LEO follower 故障follower 发生故障后会被临时踢出 ISR待该 follower 恢复后 follower 会读取本地磁盘记录的上次的 HW并将 log 文件高于 HW 的部分截取掉从 HW 开始向 leader 进行同步。等该 follower 的 LEO 大于等于该 Partition 的 HW即 follower 追上 leader 之后就可以重新加入 ISR 了。 leader 故障leader 发生故障之后会从 ISR 中选出一个新的 leader之后为保证多个副本之间的数据一致性 其余的 follower 会先将各自的 log 文件高于 HW 的部分截掉然后从新的 leader同步数据如果少于 leader 中的数据则会从 leader 中进行同步。
生产者ack机制
对于某些不太重要的数据对数据的可靠性要求不是很高能够容忍数据的少量丢失所以没必要等 ISR 中的 follower 全部接收成功。kafka为生产者提供了三种ack可靠性级别配置
0producer 不等待 broker 的 ack这一操作提供了一个最低的延迟broker 一接收到还没有写入磁盘就已经返回当 broker 故障时有可能丢失数据1 producer 等待 broker 的 ack partition 的 leader 落盘成功后返回 ack如果在 follower同步成功之前 leader 故障那么将会丢失数据-1/allproducer 等待 broker 的 ack ISR队列中 partition 的 leader 和 ISR 的follower 全部落盘成功后才返回 ack。但是如果在 follower 同步完成后broker 发送 ack 之前 leader 发生故障那么会造成数据重复
ExactlyOnce
将服务器的 ACK 级别设置为-1可以保证 Producer 到 Server 之间不会丢失数据即 At Least Once 语义至少发送一次相对的将服务器 ACK 级别设置为 0可以保证生产者每条消息只会被发送一次即 At Most Once 语义至多发送一次。
至少发送一次可以保证数据不丢失但是不能保证数据不重复相对的至多发送一次可以保证数据不重复但是不能保证数据不丢失。 但是对于一些非常重要的信息比如说交易数据下游数据消费者要求数据既不重复也不丢失即 Exactly Once 语义准确发送一次。
在0.11 版本的 Kafka引入了一项重大特性幂等性。所谓的幂等性就是指生产者不论向 Server 发送多少次重复数据 Server 端都只会持久化一条。幂等性结合 At Least Once 语义就构成了 Kafka 的 Exactly Once 语义。
要启用幂等性只需要将 Producer 的参数中 enable.idempotence 设置为 true 即可。
Kafka的幂等性实现其实就是将原来下游需要做的去重放在了数据上游。开启幂等性的 Producer 在初始化的时候会被分配一个 PID发往同一 Partition 的消息会附带 Sequence Number。而Broker 端会对PID, Partition, SeqNumber做缓存当具有相同主键的消息提交时 Broker 只会持久化一条。但是 PID 重启就会变化同时不同的 Partition 也具有不同主键所以幂等性无法保证跨分区跨会话的 Exactly Once。
生产者发送消息流程
kafka的生产者发送消息采用的是异步发送的方式。在消息发送的过程中涉及到了两个线程——main 线程和 Sender 线程以及一个线程共享变量——RecordAccumulator。 在生产者发送消息时main线程将消息发送给 RecordAccumulator当数据积累到 batch.size 之后sender线程才会不断从 RecordAccumulator 中拉取消息发送到 kafka的broker如果数据迟迟未达到 batch.sizesender 线程等待 linger.time 之后就会发送数据。
消费者
消费者分区分配策略
一个消费者组中有多个消费者一个主题有多个分区所以必然会涉及到分区的分配问题即确定哪个个分区由哪个消费者来消费。
消费分配策略
round-robin会采用轮询的方式将当前所有的分区依次分配给所有的消费者range首先会计算每个消费者可以消费的分区个数然后按照顺序将指定个数范围的分区分配给各个消费者sticky这种分区策略是最新版本中新增的一种策略将现有的分区尽可能均衡的分配给各个消费者存在此目的的原因在于Round-Robin和Range分配策略实际上都会导致某几个消费者承载过多的分区从而导致消费压力不均衡
轮询就不必说了就是把分区按照hash排序然后分配。range按范围分配先将所有的分区放到一起然后排序按照平均分配的方式计算每个消费者会得到多少个分区如果没有除尽则会将多出来的分区依次计算到前面几个消费者。 比如这里是三个分区和两个消费者那么每个消费者至少会得到1个分区而3除以2后还余1那么就会将多余的部分依次算到前面几个消费者也就是这里的1会分配给第一个消费者
如果按照Range分区方式进行分配其本质上是依次遍历每个topic然后将这些topic的分区按照其所订阅的消费者数量进行平均的范围分配。这种方式从计算原理上就会导致排序在前面的消费者分配到更多的分区从而导致各个消费者的压力不均衡。
消费者消费数据问题
消费者在消费的时候需要维护一个offset用于记录消费的位置当offset提交时会有两个问题重复消费和漏消费
当提交的offset小于当前程序处理的最后一条消息的offset,会造成重复消费。情景先消费后提交offset,如果消费成功、提交失败消费者下次获取的offset还是以前的所以会造成重复消费。当提交的offset大于当前程序处理的最后一条消息的offset,会造成漏消费。情景先提交offset后消费,如果提交成功、消费失败消费者下次获取的offset已经是新的所以会造成漏消费。
这里就需要注意offset的提交方式offset默认是自动提交当然这会造成消费的不准确。offset提交方式
异步提交当前offset同步提交当前offset异步提交当前offset同步和异步组合提交提交指定的offset
建议将offset保存在数据库中使当前业务与offset提交绑定起来这样可以一定程度避免重复消费问题重复消费的问题一方面需要消息中间件来进行保证。另一方面需要自己的处理逻辑来保证消息的幂等性。
kafka事务
kafka从0.11版本开始引入了事务支持。事务可以保证kafka在Exactly Once语义的基础上生产和消费可以跨分区和会话要么全部成功要么全部失败。
生产者为了实现跨分区跨会话的事务需要引入一个全局唯一的事务ID并将生产者获得的PID和事务ID绑定。这样当生产者重启后就可以通过正在进行的事务ID获得原来的PID进而保证精准发送一次 为了管理 Transaction Kafka 引入了一个新的组件 Transaction Coordinator。 Producer 就是通过和 Transaction Coordinator 交互获得 Transaction ID 对应的任务状态。 Transaction Coordinator 还负责将事务所有写入 Kafka 的一个内部 Topic这样即使整个服务重启由于事务状态得到保存进行中的事务状态可以得到恢复从而继续进行。 消费者对于消费者而言事务的保证就会相对较弱尤其时无法保证提交的信息被精确消费。这是由于消费者可以通过offset访问任意信息而且不同的Segment File生命周期不同同一事务的消息可能会出现重启后被删除的情况
消息积压
如果线上遇到大量消息积压,那就是线上故障了.最可能是消费者出现故障
一般这个时候只能临时紧急扩容了:
先修复消费者的问题确保其恢复消费速度然后将现有消费者都停掉新建一个 topicpartition 是原来的 10 倍临时建立好原先 10 倍的 queue 数量;然后写一个临时的分发数据的 consumer 程序这个程序部署上去消费积压的数据消费之后不做耗时的处理直接均匀轮询写入临时建立好的 10 倍数量的 queue;接着临时征用 10 倍的机器来部署 consumer每一批 consumer 消费一个临时 queue 的数据。这种做法相当于是临时将 queue 资源和 consumer 资源扩大 10 倍以正常的 10 倍速度来消费数据;等快速消费完积压数据之后得恢复原先部署的架构重新用原先的 consumer 机器来消费消息;
如果没有消费者没有出现问题,而出现了消费积压的情况可以参考以下思路:
提高消费并行度批量方式消费跳过非重要方式消费优化消息消费业务处理过程,简化过程
