大型地方门户网站源码,wordpress rest api开发,网站做百度小程序改造的好处,问答系统网站建设目录
1. 为什么使用消息队列
2. 消息队列的缺点
3. 消息队列如何选型#xff1f;
4. 如何保证消息队列是高可用的
5. 如何保证消息不被重复消费#xff08;见第二条#xff09;
6. 如何保证消息的可靠性传输#xff1f;
7. 如何保证消息的顺序性#xff08;即消息幂…目录
1. 为什么使用消息队列
2. 消息队列的缺点
3. 消息队列如何选型
4. 如何保证消息队列是高可用的
5. 如何保证消息不被重复消费见第二条
6. 如何保证消息的可靠性传输
7. 如何保证消息的顺序性即消息幂等性
8. 如何快速处理积压消息
9. MQ的架构和重要组件RocketMQ
10. RabbitMQ运转流程 1. 为什么使用消息队列 解耦 一旦A挂了就会导致下游所有系统都没有数据使用消息中间件来解耦A只需要把数据发给中间 件下游系统自行调用 异步系统C发送个消息到MQ中间件里由系统D消费到消息之后慢慢的异步来执行这个耗 时2s的业务处理。这种方式直接将核心链路的执行性能提升了10倍。 削峰某套系统低峰期并发也就100多个高峰期并发量会增到5000以上而这个数据库只 支撑每秒1000左右的并发写入一旦高峰期出现并发量很大就容易宕机。使用MQ后消息 被MQ保存起来然后系统按照自己的消费能力来消费比如每秒消费1000个这样慢慢写入 数据库就不会轻易宕机。 2. 消息队列的缺点 可用性低本来其他系统只要运行好好的那你的系统就是正常的。现在你非要加入个消息队 列进去那消息队列挂了你的系统不是挂了。因此系统可用性会降低 。解决方法使用MQ集群防止一个挂了还能继续使用 复杂性高需要考虑消息有没有重复消费即消息幂等性、消息丢失、保证消息传递性等问 题 为什么会出现这种情况 正常情况下消费者在消费完消息后会发送一个确认的消息给消息队列消息队列知道该消息被消费了就会把该消息从队列删除。RabbitMQ是发送一个ACK确认消息RocketMQ是返回一个CONSUME_SUCESS。 解决方法 如果消息是数据库插入操作给消息一个唯一主键如果出现重复消费的情况会导致主键冲突数据库避免有脏数据。如果消息是set入redis的话是不会存在重复的因set操作是幂等操作。 如果上面两种情况还不行上大招。准备一个第三方介质来做消费记录。以redis为例给消息分配一个全局id只要消费过该消息将id,message以K-V形式写入redis. 那消费者开始消费前先去redis中查询有没有消费记录即可。 数据一致性问题MQ带来系统响应提高但是如果消息没消费就会导致数据不统一 解决即防止生产者弄丢数据、消息队列弄丢数据、消费者弄丢数据。消息队列一般会 持久化到磁盘生产者数据丢失MQ事务会回滚可以尝试重新发送。消费者丢失数据一 般都是采用了自动确认消息模式导致消费信息被删只要改为手动即可即消费者消费 完之后调用一个MQ的确认方法就行了。 缺点rabbitMQ事务开启就会变为同步阻塞操作生产者会阻塞等待是否发送成功 太耗性能会造成吞吐量的下降。 消息队列的延时与过期失效问题 解决消息队列的延迟和过期失效是消息队列的自我保护机制目的是为了防止本身被 挤爆当然也可以关闭这个机制。但是不推荐关闭可以改为某个消息消费失败5次后 可以把这个消息丢弃等但是丢弃的数据最好写个临时程序又重新放入MQ队列或记录下 来重新处理。 3. 消息队列如何选型 更新频率 RocketMQ 比 RabbitMQ 频繁比 ActiveMQ 频繁 中小公司建议用RabbitMQ 大公司用 ActiveMQ 大数据用 Kafka 4. 如何保证消息队列是高可用的 1 RabbitMQ 中有三种模式单机模式、普通集群模式、镜像集群模式。生产没人用单机模式 普通集群模式默认的集群模式对于Queue来说消息实体只存在于其中一个节点A、B两个 节点仅有相同的元数据即队列结构但队列的元数据仅保存有一份即创建该队列的rabbitmq 节点A节点当A节点宕机你可以去其B节点查看发现该队列已经丢失。 当消息进入A节点的Queue中后consumer从B节点拉取时RabbitMQ会临时在A、B间进行消息 传输把A中的消息实体取出并经过B发送给consumer所以consumer应平均连接每一个节点 从中取消息。该模式存在一个问题就是当A节点故障后B节点无法取到A节点中还未消费的消息实 体。如果做了队列持久化或消息持久化那么得等A节点恢复然后才可被消费并且在A节点恢复 之前其它节点不能再创建A节点已经创建过的持久队列如果没有持久化的话消息就会失丢。这 种模式更适合非持久化队列只有该队列是非持久的客户端才能重新连接到集群里的其他节点 并重新创建队列。假如该队列是持久化的那么唯一办法是将故障节点恢复起来。 镜像模式把需要的队列做成镜像队列存在于多个节点。其实质和普通模式不同之处在于消息 实体会主动在镜像节点间同步而不是在consumer取数据时临时拉取。该模式带来的副作用也很 明显除了降低系统性能外如果镜像队列数量过多加之大量的消息进入集群内部的网络带宽 将会被这种同步通讯大大消耗掉。所以在对可靠性要求较高的场合中适用 主备模式主节点提供读写备用节点不提供读写。如果主节点挂了就切换到备用节点原来的 备用节点升级为主节点提供读写服务当原来的主节点恢复运行后原来的主节点就变成备用节点 2 RocketMQ 集群 多Master模式一个集群无Slave全是Master例如2个Master或者3个Master这种模式的优 缺点如下 优点配置简单单个Master宕机或重启维护对应用无影响在磁盘配置为RAID10时即使 机器宕机不可恢复情况下由于RAID10磁盘非常可靠消息也不会丢异步刷盘丢失少量消 息同步刷盘一条不丢性能最高缺点单台机器宕机期间这台机器上未被消费的消息在机器恢复之前不可订阅消息实时性 会受到影响。 多Master多Slave模式异步复制每个 Master 配置一个 Slave有多对Master-Slave HA 采用异步复制方式主备有短暂消息延迟毫秒级。 优点即使磁盘损坏消息丢失的非常少且消息实时性不会受影响因为Master 宕机后 消费者仍然可以从 Slave消费此过程对应用透明。不需要人工干预。性能同多 Master 模式 几乎一样。 缺点 Master 宕机磁盘损坏情况会丢失少量消息。 多Master多Slave模式同步双写线上使用的话推荐使用此模式集群每个 Master 配置一个 Slave有多对Master-Slave HA采用同步双写方式主备都写成功向应用返回成功。 优点数据与服务都无单点 Master宕机情况下消息无延迟服务可用性与数据可用性都 非常高 缺点性能比异步复制模式略低大约低 10%左右发送单个消息的 RT会略高。目前主宕机后备机不能自动切换为主机。 Dledger部署RocketMQ 4.5以后出现每个Master配置二个 Slave 组成 Group可以有多个 Dledger Group一但 Master 宕机Dledger 就可以从剩下的两个 Broker 中选举一个 Master 继 续对外提供服务。使用raft算法进行选举。 5. 如何保证消息不被重复消费见第二条 6. 如何保证消息的可靠性传输 在使用消息队列过程中应该做到消息不能多消费也不能少消费因此可靠性传输从以下三个角度分析RocketMQ 生产者弄丢数据对RocketMQ来说设计之初就避免了生产者丢失消息即事务消息 事务消息是在分布式系统中保证最终一致性的2PC(两阶段提交)的消息实现。他可以保证本地 事务执行与消息发送两个操作的原子性也就是这两个操作一起成功或者一起失败。 事务消息只保证消息发送者的本地事务与发消息这两个操作的原子性因此事务消息的示例 只涉及到消息发送者对于消息消费者来说并没有什么特别的。 流程先发送一个half消息(可以理解为半个消息这个消息作用在事务生效前检查一rocketMQ是否活着对下游消费者不可见上图步骤1判断rocketMQ的状态然后rocketMQ回复给生产者(步骤2)生产者再去操作自己的事务(步骤3)如果操作事务失败给rocketMQ返回一个本地事务的状态rollback然后rocketMQ丢弃掉。如果操作成功给rocketMQ返回一个commit给下游服务(步骤4这返回有三个状态分别是commit、rollback、unknow)。 如果生产者在处理本地事务过于太久先给rocketMQ返回一个unkown步骤4那 rocketMQ就不会跟消费者任何接触而rocketMQ过会给生产者一个回查状态(步骤5)来确定 你到底执行完没(步骤6)返回给rocketMQ(步骤7这里返回的状态还是commit、rollback、 unknow)。返回过来的commit就给消费者返回是rollback就丢弃返回的unknow的话就会等待下一次去回查以此类推循环查(默认是15次)。 缺点 因为要发个half消息就会比普通消息慢而这只是保证消息发送端的正常发送 使用限制事务消息不支持延迟消息和批量消息 为了避免单个消息被检查确认多次而导致消息堆积我们默认单个消息检查次数为 15次。这个次数可以该broker配置文件的transactionCheckMax属性 在broker配置文件的transactionMsgTimeout来设置特定时间长度之后被检查 事务消息只保证了发送者的本地事务和发送消息这两个操作的原子性但是并不保 证消费者本地事务的原子性 RocketMQ内部保证消息不丢失使用同步刷盘DledgerrocketMQ 4.7版本才有来避免类似问题 刷盘机制 同步刷盘节点收到消息后将数据持久化到硬盘后再返回成功 异步刷盘节点收到消息后将消息存储在内存中先返回成功再持久化到硬盘 中。 Dledger它是通过两段式提交的方式保证文件在主从之间成功同步 Dledger的两段式(uncommitted阶段和commited阶段)内容如下 Leader Broker上的Dledger收到一条数据后会标记为uncommitted状态然后 他通过自己的DledgerServer组件把这个uncommitted数据发给Follower Broker的 DledgerServer组件。 接着Follower Broker的DledgerServer收到uncommitted消息之后必须返回一个 ack给Leader Broker的Dledger。然后如果Leader Broker收到超过半数的 Follower Broker返回的ack之后就会把消息标记为committed状态。 再接下来Leader Broker上的DledgerServer就会发送committed消息给Follower Broker上的DledgerServer让他们把消息也标记为committed状态。这样就基 于Raft协议完成了两阶段的数据同步。 消费者保证消息不丢失 要保证消息不丢失消费端就不要使用异步消费机制。因为异步消费可能存在消费者本地出现 问题而无法再去告知RocketMQ应该消费端进行消费结束后在返回给RocketMQ一个ack 确认消息正常消费 NameServer集群全部挂了保证消息不丢失NameServer集群挂了生产者无法往新的Topic发送消息只能设计一个降级方案来处理(可 以先把消息存在数据库或redis等RocketMQ恢复后再发送出去等方案) 保证消息不丢失主要的总结 生产者使用事务消息机制。Broker配置同步刷盘Dledger主从架构消费者不要使用异步消费。整个MQ挂了之后准备降级方案 缺点以上消息零丢失方案在各个环节都大量的降低了系统的处理性能以及吞吐量。在很多场景下这套方案带来的性能损失的代价可能远远大于部分消息丢失的代价。 所以我们在设计RocketMQ使用方案时要根据实际的业务情况来考虑。 例如如果针对所有服务器都在同一个机房的场景完全可以把Broker配置成异步刷盘来提 升吞吐量。而在有些对消息可靠性要求没有那么高的场景在生产者端就可以采用其他一些更 简单的方案来提升吞吐而采用定时对账、补偿的机制来提高消息的可靠性。而如果消费者不 需要进行消息存盘那使用异步消费的机制带来的性能提升也是非常显著的。 RabbitMQ 生产者丢数据从生产者弄丢数据这个角度来看RabbitMQ提供transaction和confirm模式 来确保生产者不丢消息 transaction机制就是说发送消息前开启事务channel.txSelect(),然后发送消息如果 发送过程中出现什么异常事务就会回滚channel.txRollback(),如果发送成功则提交事务 channel.txCommit。这种方式有个缺点吞吐量下降。 生产上用confirm模式的居多。一旦channel进入confirm模式所有在该信道上发布的消息都 将会被指派一个唯一的ID从1开始一旦消息被投递到所有匹配的队列之后rabbitMQ 就会发送一个ACK给生产者包含消息的唯一ID这就使得生产者知道消息已经正确到达目 的队列了。如果rabbitMQ没能处理该消息则会发送一个Nack消息给你你可以进行重试操作。但是批量发送多少确定一次是关键数量太少效率太低。数量多如果出问题是不是所有的都要重发。 消息队列丢数据 消息从Exchange路由到Queue出现问题 有两种方式处理无法路由的消息一种是让服务端重发给生产者另外一种是让交换机 路由到另外一个备份的交换机。 消息持久化到磁盘 处理消息队列丢数据的情况一般是开启持久化磁盘的配置。这个持久化配置可以和 confirm机制配合使用你可以在消息持久化磁盘后再给生产者发送一个Ack信号。这 样如果消息持久化磁盘之前rabbitMQ阵亡了那么生产者收不到Ack信号生产者 会自动重发。 使用集群保证消息一定会被写入磁盘 消费者丢数据 消费者丢数据一般是因为采用了自动确认消息模式。这种模式下消费者会自动确认收到信息 即自动发ACK给消息队列。这时rabbitMQ会立即将消息删除这种情况下如果消费者 出现异常而未能处理消息就会丢失该消息。至于解决方案采用手动确认消息即可 7. 如何保证消息的顺序性即消息幂等性 MQ 的顺序问题分为全局有序和局部有序 全局有序整个MQ系统的所有消息严格按照队列先入先出顺序进行消费。 局部有序只保证一部分关键消息的消费顺序 在RocketMQ的通常情况下发送者发送消息时会通过MessageQueue轮询的方式保证消息尽量均匀的分布到所有的MessageQueue上而消费者也就同样需要从多个MessageQueue上消费消息。而 MessageQueue是RocketMQ存储消息的最小单元他们之间的消息都是互相隔离的在这种情况下 是无法保证消息全局有序的。 而对于局部有序的要求只需要将有序的一组消息都存入同一个MessageQueue里这样 MessageQueue的FIFO设计天生就可以保证这一组消息的有序。RocketMQ中可以在发送者发送消息 时指定一个MessageSelector对象让这个对象来决定消息发入哪一个MessageQueue。这样就可以保 证一组有序的消息能够发到同一个MessageQueue里。 另外通常所谓的保证Topic全局消息有序的方式就是将Topic配置成只有一个MessageQueue队列(默 认是4个)。这样天生就能保证消息全局有序了。 8. 如何快速处理积压消息 场景例如某一天一个数据库突然挂了大家大概率就会集中处理数据库的问题。等好不容易把数 据库恢复过来了这时基于这个数据库服务的消费者程序就会积累大量的消息。 查看消息堆积去web控制台处理消息堆积 如果Topic下的MessageQueue配置得是足够多的那每个Consumer实际上会分配多个 MessageQueue来进行消费。这个时候就可以简单的通过增加Consumer的服务节点数量 来加快消息的消费等积压消息消费完了再恢复成正常情况。最极限的情况是把Consumer 的节点个数设置成跟MessageQueue的个数相同。但是如果此时再继续增加Consumer的服 务节点就没有用了。 而如果Topic下的MessageQueue配置得不够多的话那就不能用上面这种增加Consumer节 点个数的方法了。这时怎么办呢 这时如果要快速处理积压的消息可以创建一个新的 Topic配置足够多的MessageQueue。然后把所有消费者节点的目标Topic转向新的Topic 并紧急上线一组新的消费者只负责消费旧Topic中的消息并转储到新的Topic中这个速度 是可以很快的。然后在新的Topic上就可以通过增加消费者个数来提高消费速度了。之后再 根据情况恢复成正常情况。 消息设置了过期时间过期就丢了 RabbitMQ可以设置过期时间TTL如果消息在queue中积压超过一定的时间就会被rabbitmq 给清理掉这个数据就没了。 解决方法夜深人静写个程序手动查询丢失的部分数据重新补吧 积压消息长时间没处理MQ放不下了 先把积压的消息读到redis或者es快速消费掉积压的消息降低MQ的压力然后晚上重新导数据 9. MQ的架构和重要组件RocketMQ NameServer(元数据管理)NameServer是RocketMQ的寻址服务存储Broker的路由信息以及配置 信息用户端生成者消费者依靠NameServer去选择对于的Broker服务即每个Broker在启动的时 候会到NameServer注册Producer在发送消息前会根据Topic到NameServer获取到Broker的路由信 息Consumer也会定时获取Topic的路由信息类似注册中心。主要负责对于源数据的管理包括了对 于Topic和路由信息的管理。 Broker消息中转角色负责存储消息转发消息处理Producer发送消息请求Consumer消费消息的请求并且进行消息的持久化以及服务端过滤就是集群中很重的工作都是交给了Broker进行处理。 Producer(生产者) Producer Group一类Producer的集合名称这类Producer通常发送一类消息且发送逻辑一致(设计组是防止rockertmq独有的事务消息回传的时候防止生产者挂了机制)。 Topic(主题区分消息)一条消息必须有一个Topic可以看做是你的信件邮寄地址。与生产者和消费者的关系非常松散一个Topic可以有0个、1个、多个生产者向其发送消息一个生产者也可以同时向不同的Topic发送消息。比如一个电商系统的消息可以分为交易消息、物流消息。 Tag表示消息的第二级类型可以没有 Message Queue(消息队列最小单位)一个topic下我们可以设置多个queue(消息队列)。当我们发送消息时需要要指定该消息的topic。RocketMQ会轮询该topic下的所有队列将消息发送出去。Queue的引入使得消息的存储可以分布式集群化具有了水平扩展能力。每个Queue内部是有序的在RocketMQ中分为读和写两种队列一般来说读写队列数量一致如果不一致就会出现很多问题。至少要保证读队列数写队列数否则会发现一些队列中的数据读取不到。 RabbitMQ: RabbitMQ Server也叫broker server它是一种传输服务。他的角色就是维护一条从Producer到Consumer的路线保证数据能够按照指定的方式进行传输。 Exchange交换器生产者将消息发送到Exchange由Exchange将消息路由到一个或多个Queue中或者丢弃。Exchange与Queue关系是多对多。Exchange并不存储消息。RabbitMQ中的Exchange有direct、fanout、topic常见三种类型每种类型对应不同的路由规则。(用于接受、分发消息) Queue(队列):RabbitMQ的内部对象用于存储消息。消息消费者就是通过订阅队列来获取消息RabbitMQ中的消息都只能存储在Queue中生产者生产消息并最终投递到Queue中消费者可以从Queue中获取消息并消费。多个消费者可以订阅同一个Queue这时Queue中的消息会被平均分摊给多个消费者进行处理而不是每个消费者都收到所有的消息并处理。存储生产者的消息本质是一个Erlang数据库Queue与消费者是多对多的关系 Banding用于把交换器的消息绑定到队列上。是Exchange和Queue之间的虚拟连接binding里包含routing key Routing Key用于把生成者的数据分配到交换器上。可以理解为队列唯一标识(或理解为路由规则)生产者在将消息发送给Exchange的时候一般会指定一个routing key来指定这个消息的路由规则而这个routing key需要与Exchange Type及binding key联合使用才能最终生效。 Channels信道它建立在TCP连接中。数据流动都是在Channel中进行的。也就是说一般情况是程序起始建立TCP连接第二步就是建立这个Channel。(消息推送使用的通道)。几乎所有操作都在信道中进行是消息读写的通道。 VirtualHost虚拟主机权限控制的基本单位一个Virtual Host里面有若干Exchange和 MessageQueue以及指定被哪些user使用不同virtualHost相互隔离。不同的应用可以跑在不同的 vhost 中 补充 ①怎么实现消息到达队列后消费者能在第一时间获取 利用事件机制实时监听 ②生产者如何把消息发给多个队列 生产者把消息发送给交换机交换机利用绑定规则绑定多个队列队列里的消息只能消费一次如果 多个系统消费我们就需要有多个队列 ③消息是存到交换机还是队列呢本质是什么 消息是存到队列队列本质是Erlang语言写的数据库而交换机不会存储消息只是消息有个特殊的标 识消息到达交换机的时候标识就会根据绑定规则的情况把消息发到对应的一个或多个队列里。 ④交换机和队列是一对一的关系吗消费者和队列呢 交换机和队列是多对多消费者和队列也是多对多生产者和交换机也是多对多但是一般生产环境尽 量配置一对一的方便排查消息来源哪里特殊业务除外。 10. RabbitMQ运转流程 生产者发送消息的时候 生产者连接到RabbitMQ Broker 建立一个连接( Connection) 开启一个信道(Channel) 生产者声明一个交换器并设置相关属性比如交换机类型、是否持久化等 生产者声明一个队列井设置相关属性比如是否排他、是否持久化、是否自动删除等 生产者通过路由键将交换器和队列绑定起来 生产者发送消息至RabbitMQ Broker其中包含路由键、交换器等信息 相应的交换器根据接收到的路由键查找相匹配的队列。 如果找到则将从生产者发送过来的消息存入相应的队列中。 如果没有找到则根据生产者配置的属性选择丢弃还是回退给生产者 关闭信道。 关闭连接。消费者接收消息的过程: 消费者连接到RabbitMQ Broker 建立一个连接(Connection ) 开启一个信道(Channel) 。 消费者向RabbitMQ Broker 请求消费相应队列中的消息可能会设置相应的回调函数 以及做一些准备工作 等待RabbitMQ Broker 回应并投递相应队列中的消息 消费者接收消息。 消费者确认( ack) 接收到的消息。 RabbitMQ 从队列中删除相应己经被确认的消息。 关闭信道。 关闭连接
