沂源手机网站建设公司,百度商桥怎么嵌入网站,服务行业网站建设,鞋网站模版MQ的模式#xff1a;
基本消息模式#xff1a;一个生产者#xff0c;一个消费者work模式#xff1a;一个生产者#xff0c;多个消费者订阅模式#xff1a; fanout广播模式#xff1a;在Fanout模式中#xff0c;一条消息#xff0c;会被所有订阅的队列都消费。 在广播… MQ的模式
基本消息模式一个生产者一个消费者work模式一个生产者多个消费者订阅模式 fanout广播模式在Fanout模式中一条消息会被所有订阅的队列都消费。 在广播模式下消息发送流程
可以有多个消费者队列的消费者都能拿到消息。实现一条消息被多个消费者消费交换机把消息发送给绑定过的所有队列生产者发送的消息只能发送到交换机交换机来决定要发给哪个队列生产者无法决定。每个队列都要绑定到Exchange交换机每个消费者有自己的queue队列 direct定向模式 在Direct下 队列与交换机的绑定不能是任意绑定了而是要指定一个RoutingKey路由key消息的发送方在 向 Exchange发送消息时也必须指定消息的 RoutingKey。Exchange不再把消息交给每一个绑定的队列而是根据消息的Routing Key进行判断只有队列的Routingkey与消息的 Routing key完全一致才会接收到消息 topic通配符模式
Topic类型的Exchange与Direct相比都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定Routing key 的时候使用通配符 Routingkey 一般都是有一个或多个单词组成多个单词之间以”.”分割例如 item.insert
通配符规则
#匹配一个或多个词
*匹配不多不少恰好1个词
举例 item.#能够匹配item.spu.insert 或者 item.spu item.*只能匹配item.spu RabbitMQ 使用信道的方式来传输数据。信道是建立在真实的 TCP 连接内的虚拟连接且每条 TCP 连接上的信道数量没有限制。
单一模式集群普通模式如果做了持久化需要等待rabbit01恢复才可以被消费如果没有做持久化可能造成消息丢失镜像模式
1、什么是RabbitMQ为什么使用RabbitMQ主要特性
答RabbitMQ是一款开源的Erlang编写的基于AMQP协议的消息中间件
可以用它来解耦、异步、削峰添谷。
主要特性
可伸缩性集群服务【主从模式的】消息持久化从内存持久化消息到硬盘再从硬盘加载到内存**【队列的持久化、交换机的持久化、消息持久化】**
1.1、解耦为面向服务的架构SOA提供基本的最终一致性实现
场景说明用户下单后订单系统需要通知库存系统。传统的做法是订单系统调用库存系统的接口。
传统模式的缺点
假如库存系统无法访问则订单减库存将失败从而导致订单失败订单系统与库存系统耦合
引入消息队列
订单系统用户下单后订单系统完成持久化处理将消息写入消息队列返回用户订单下单成功库存系统订阅下单的消息采用拉/推的方式获取下单信息库存系统根据下单信息进行库存操作假如在下单时库存系统不能正常使用。也不影响正常下单因为下单后订单系统写入消息队列就不再关心其他的后续操作了。实现订单系统与库存系统的应用解耦为了保证库存肯定有可以将队列大小设置成库存数量或者采用其他方式解决。
基于消息的模型关心的是“通知”而非“处理”。
短信、邮件通知、缓存刷新等操作使用消息队列进行通知。
1.2、消息队列和RPC的区别与比较
RPC: 异步调用及时获得调用结果具有强一致性结果关心业务调用处理结果。
消息队列两次异步RPC调用将调用内容在队列中进行转储并选择合适的时机进行投递错峰流控
异步提升效率
场景说明用户注册后需要发注册邮件和注册短信。传统的做法有两种
1.串行的方式2.并行方式
1串行方式将注册信息写入数据库成功后发送注册邮件再发送注册短信。以上三个任务全部完成后返回给客户端
2并行方式将注册信息写入数据库成功后发送注册邮件的同时发送注册短信。以上三个任务完成后返回给客户端。与串行的差别是并行的方式可以提高处理的时间
3引入消息队列将不是必须的业务逻辑异步处理。
流量削峰
流量削锋也是消息队列中的常用场景一般在秒杀或团抢活动中使用广泛。关于秒杀系列文章可以关注公众号Java技术栈获取阅读。
应用场景系统其他时间A系统每秒请求量就100个系统可以稳定运行。系统每天晚间八点有秒杀活动每秒并发请求量增至1万条但是系统最大的处理能力只能每秒处理1000个请求于是系统崩溃服务器宕机。
之前架构大量用户100万用户通过浏览器在晚上八点高峰期同时参与秒杀活动。大量的请求涌入我们的系统中高峰期达到每秒钟5000个请求大量的请求打到MySQL上每秒钟预计执行3000条SQL。
但是一般的MySQL每秒钟扛住2000个请求就不错了如果达到3000个请求的话可能MySQL直接就瘫痪了从而系统无法被使用。但是高峰期过了之后就成了低峰期可能也就1万用户访问系统每秒的请求数量也就50个左右整个系统几乎没有任何压力。
引入MQ100万用户在高峰期的时候每秒请求有5000个请求左右将这5000请求写入MQ里面系统A每秒最多只能处理2000请求因为MySQL每秒只能处理2000个请求。
系统A从MQ中慢慢拉取请求每秒就拉取2000个请求不要超过自己每秒能处理的请求数量即可。MQ每秒5000个请求进来结果只有2000个请求出去所以在秒杀期间将近一小时可能会有几十万或者几百万的请求积压在MQ中。
这个短暂的高峰期积压是没问题的因为高峰期过了之后每秒就只有50个请求进入MQ了但是系统还是按照每秒2000个请求的速度在处理所以说只要高峰期一过系统就会快速将积压的消息消费掉。
我们在此计算一下每秒在MQ积压3000条消息1分钟会积压18万1小时积压1000万条消息高峰期过后1个多小时就可以将积压的1000万消息消费掉。 2、RabbitMQ有什么优缺点
答优点解耦、异步、削峰
缺点降低了系统的稳定性本来系统运行好好的现在你非要加入个消息队列进去那消息队列挂了系统也就崩溃了。因此系统可用性会降低
增加了系统的复杂性加入了消息队列要多考虑很多方面的问题比如一致性问题、如何保证消息不被重复消费、如何保证消息可靠性传输等。因此需要考虑的东西更多复杂性增大。 3、如何保证RabbitMQ的高可用
答搭建集群的RabbitMQ服务器提供服务一台风险太大
4、如何保证RabbitMQ不被重复消费
答先说为什么会重复消费正常情况下消费者在消费消息的时候消费完毕后会发送一个确认消息给消息队列消息队列就知道该消息被消费了就会将该消息从消息队列中删除 但是因为网络传输等等故障确认信息没有传送到消息队列导致消息队列不知道自己已经消费过该消息了再次将消息分发给其他的消费者。
【即消费者发送 ACK 前崩溃MQ 会重新投递消息可能导致重复处理】
针对以上问题一个解决思路是保证消息的唯一性就算是多次传输不要让消息的多次消费带来影响保证消息幂等性
比如在写入消息队列的数据做唯一标示比如放到redis中消费消息时根据唯一标识判断是否消费过
ACK两种模式
自动 ACKAuto-Acknowledgement
定义消费者收到消息后MQ 立即自动标记消息为已确认无论消费者是否处理成功。特点 简单易用但可靠性低。若消费者处理消息时崩溃消息会丢失因为 MQ 已认为消息被确认。
适用场景允许少量消息丢失的非关键业务如日志采集。
手动 ACKManual Acknowledgement
定义消费者在处理消息后 显式发送 ACKMQ 才会删除消息。特点 可靠性高但需开发者主动控制。若处理失败可发送 NACKNegative Acknowledgement拒绝消息触发重试或进入死信队列。
适用场景金融交易、订单处理等对可靠性要求高的场景。
5、如何保证RabbitMQ消息的可靠传输(即保证消息不会丢失)
答消息不可靠的情况可能是消息丢失劫持等原因
丢失又分为生产者丢失消息、消息队列丢失消息、消费者丢失消息
生产者丢失消息【生产者发消息的可靠性】
从生产者弄丢数据这个角度来看RabbitMQ提供transaction和confirm模式来确保生产者不丢消息
transaction机制就是说发送消息前开启事务channel.txSelect(),然后发送消息如果发送过程中出现什么异常事务就会回滚channel.txRollback(),如果发送成功则提交事务channel.txCommit()。然而这种方式有个缺点吞吐量下降
confirm模式用的居多一旦channel进入confirm模式所有在该信道上发布的消息都将会被指派一个唯一的ID从1开始一旦消息被投递到所有匹配的队列之后
MQ就会发送一个ACK给生产者包含消息的唯一ID让生产者知道消息已经正确到达目的队列了
如果MQ没能处理该消息则会发送一个Nack消息给生产者让生产者进行重试操作。
处理Ack和Nack的代码如下所示
rabbitmq:host: 192.168.112.129port: 5672username: adminpassword: pass#确认消息已发送到交换机(Exchange)#确认消息已发送到队列(Queue)publisher-confirms: truepublisher-returns: true
消息队列丢数据消息持久化。【消息队列数据的可靠性】
处理消息队列丢数据的情况一般是开启持久化磁盘的配置。
这个持久化配置可以和confirm机制配合使用你可以在消息持久化磁盘后再给生产者发送一个Ack信号。
这样如果消息持久化磁盘之前rabbitMQ阵亡了那么生产者收不到Ack信号生产者会自动重发。
那么如何持久化呢
这里顺便说一下吧其实也很容易就下面两步declare/dɪˈkler/
队列持久化将queue的持久化标识durable(/ˈdʊrəbl/)设置为true则代表是一个持久的队列 交换机持久化 消息持久化发送消息的时候将delivery_mode2// 设置消息是否持久化1 非持久化 2持久化 这样设置以后即使rabbitMQ挂了重启后也能恢复数据 消费者丢失消息【消费者消费数据的可靠性】
消费者丢数据一般是因为采用了自动确认消息模式改为手动确认消息即可 消费者在收到消息之后处理消息之前会自动回复RabbitMQ已收到消息
如果这时处理消息失败就会丢失该消息
解决方案消费者接收每一条消息后都必须进行确认消息接收和消息确认是两个不同操作。只有消费者确认了消息RabbitMQ 才能安全地把消息从队列中删除。
交换机、队列、消息持久化方式总结
为了提高并发能力MQ的数据默认是在内存中存储的包括交换机、队列、消息。
这样就会出现数据安全问题如果服务宕机存储在MQ中未被消费的消息都会丢失。
所以我们需要将交换机、队列、消息持久化到硬盘以防服务宕机。
交换机持久化 队列持久化 消息持久化 6、如何保证RabbitMQ消息的顺序性
答单线程消费保证消息的顺序性对消息进行编号消费者处理消息是根据编号处理消息
把需要顺序性消费的消息放到同一个队列里只允许一个消费者去消费这个队列不需要保证顺序的就放到其他队列里。 7、如何确保消息正确地发送至 RabbitMQ 如何确保消息接收方消费了消息
发送方确认模式
将信道设置成 confirm 模式发送方确认模式则所有在信道上发布的消息都会被指派一个唯一的 ID。
一旦消息被投递到目的队列后或者消息被写入磁盘后可持久化的消息信道会发送一个确认给生产者包含消息唯一 ID。
如果 RabbitMQ 发生内部错误从而导致消息丢失会发送一条 nacknotacknowledged未确认消息。
发送方确认模式是异步的生产者应用程序在等待确认的同时可以继续发送消息。当确认消息到达生产者应用程序生产者应用程序的回调方法就会被触发来处理确认消息。
接收方确认机制
消费者接收每一条消息后都必须进行确认消息接收和消息确认是两个不同操作。只有消费者确认了消息RabbitMQ 才能安全地把消息从队列中删除。
这里并没有用到超时机制RabbitMQ 仅通过 Consumer 的连接中断来确认是否需要重新发送消息。也就是说只要连接不中断RabbitMQ 给了 Consumer 足够长的时间来处理消息。保证数据的最终一致性
下面罗列几种特殊情况
如果消费者接收到消息在确认之前断开了连接或取消订阅RabbitMQ 会认为消息没有被分发然后重新分发给下一个订阅的消费者。可能存在消息重复消费的隐患需要去重
如果消费者接收到消息却没有确认消息连接也未断开则 RabbitMQ 认为该消费者繁忙将不会给该消费者分发更多的消息。
如何解决消息队列的延时以及过期失效问题消息队列满了以后该怎么处理有几百万消息持续积压几小时说说怎么解决
消息积压处理办法临时紧急扩容
先修复 consumer 的问题确保其恢复消费速度然后将现有 cnosumer 都停掉。
新建一个 topicpartition 是原来的 10 倍临时建立好原先 10 倍的 queue 数量。
然后写一个临时的分发数据的 consumer 程序这个程序部署上去消费积压的数据消费之后不做耗时的处理直接均匀轮询写入临时建立好的 10 倍数量的 queue。
接着临时征用 10 倍的机器来部署 consumer每一批 consumer 消费一个临时 queue 的数据。这种做法相当于是临时将 queue 资源和 consumer 资源扩大 10 倍以正常的 10 倍速度来消费数据。
等快速消费完积压数据之后得恢复原先部署的架构重新用原先的 consumer 机器来消费消息。
MQ中消息失效
假设你用的是 RabbitMQRabbtiMQ 是可以设置过期时间的也就是 TTL。如果消息在 queue 中积压超过一定的时间就会被 RabbitMQ 给清理掉这个数据就没了。那这就是第二个坑了。这就不是说数据会大量积压在 mq 里而是大量的数据会直接搞丢。
我们可以采取一个方案就是批量重导这个我们之前线上也有类似的场景干过。就是大量积压的时候我们当时就直接丢弃数据了然后等过了高峰期以后比如大家一起喝咖啡熬夜到晚上12点以后用户都睡觉了。这个时候我们就开始写程序将丢失的那批数据写个临时程序一点一点的查出来然后重新灌入 mq 里面去把白天丢的数据给他补回来。也只能是这样了。假设 1 万个订单积压在 mq 里面没有处理其中 1000 个订单都丢了你只能手动写程序把那 1000 个订单给查出来手动发到 mq 里去再补一次。
mq消息队列块满了如果消息积压在 mq 里你很长时间都没有处理掉此时导致 mq 都快写满了咋办这个还有别的办法吗没有谁让你第一个方案执行的太慢了你临时写程序接入数据来消费消费一个丢弃一个都不要了快速消费掉所有的消息。然后走第二个方案到了晚上再补数据吧。
如何保证消息的队列的顺序型
在生产者发送消息给队列时我们通过绑定多个队列且每个队列都绑定唯一的消费者消费者在消费队列中数据时是按顺序来的。
即生产者将消息存放到同一个队列同一个消费者在读取该队列中的消息是按顺序读取的。 消息重试【补充】机制死信队列Dead-Letter Queue, DLX
一、死信队列的定义与作用
死信队列DLX 是消息队列系统中用于存储无法被正常消费的消息的特殊队列。这些消息被称为 死信Dead-Letter Messages通常是由于以下原因导致无法处理
消息被消费者拒绝Reject/Nack且未重新入队。消息过期Time-To-Live, TTL 超时。队列达到最大长度限制新消息被丢弃或旧消息被移除。消息无法被正确路由到目标队列如交换机绑定错误。
核心作用
容错处理避免因消息处理失败导致的无限重试或丢失。故障排查集中管理异常消息便于后续分析原因。灵活路由支持对死信消息的再处理如人工干预或自动重试。 二、消息成为死信的条件 触发条件 说明 消息被拒绝且不重入队 消费者调用 basic.reject 或 basic.nack 并设置 requeuefalse。 消息过期TTL 消息或队列设置 TTL 超时后未被消费。 队列满 队列达到最大长度x-max-length或 最大容量x-max-length-bytes。 路由失败 消息无法被交换机路由到任何队列 需配置备用交换机 alternate-exchange。 重试机制的设置
RabbitMQ自动补偿机制触发:(多用于调用第三方接口)
当我们的消费者在处理我们的消息的时候,程序抛出异常情况下触发自动补偿(默认无限次数重试)应该对我们的消息重试设置间隔重试时间,
比如消费失败最多只能重试5次,间隔3秒(防止重复消费,幂等问题)如果重试后还未消费默认丢弃如果配置了死信队列会发送到死信队列中
listener:simple:#手动签收acknowledge-mode: manual#并发消费者初始化值concurrency: 10#并发消费者的最大值max-concurrency: 20#每个消费者每次监听时可拉取处理的消息数量prefetch: 5retry:#是否开启消费者重试为false时关闭消费者重试这时消费端代码异常会一直重复收到消息enabled: true#最大重试次数max-attempts: 2#重试间隔时间单位毫秒initial-interval: 5000#重试最大时间间隔单位毫秒:当前时间间隔max-interval重试最大时间间隔max-interval: 1200000#应用于前一重试间隔的乘法器:当前时间间隔上次重试间隔*multiplier。multiplier: 2
8、消息队列的使用场景 解耦系统、异步处理、削峰填谷
9、消息的重发补充策略
一、消息重发机制
消息重发通常由消费者处理失败触发分为 自动重试 和 手动重试 两种模式。
1. 自动重试策略
触发条件 消费者抛出异常或返回失败状态。消息处理超时未在指定时间内完成。
核心配置 最大重试次数限制重试次数避免无限循环如3次。重试间隔逐步增加等待时间如指数退避。重试队列隔离将重试消息路由到独立队列避免阻塞正常消费。
2. 手动重试策略
适用场景需要人工介入的复杂错误如数据修复后重试。实现方式 将失败消息持久化到数据库或死信队列。提供管理界面手动触发重试。
二、补充策略
当自动重试无法解决问题时需结合补充策略确保最终处理。
1. 死信队列DLX兜底
功能收集所有重试失败的消息避免消息丢失。处理方式 监控告警检测死信队列堆积通知运维人员。人工处理修复数据或配置后手动重新投递消息。自动补偿通过定时任务扫描死信队列重新投递需限制频率。
2. 异步补偿任务
场景消息彻底无法处理时启动补偿流程。实现 定时任务扫描检查消息处理状态重新投递或回滚业务。事务反向操作如订单支付失败后补偿释放库存。
3. 幂等性设计
必要性重发可能导致消息重复需确保业务逻辑幂等。实现方式 唯一标识为消息分配全局唯一ID如UUID处理前检查状态。数据库约束利用唯一索引或乐观锁防止重复提交。状态机限制业务操作的状态流转路径如订单只能支付一次。
四、最佳实践
分级重试策略 首次失败立即重试网络抖动。后续重试逐步增加间隔指数退避。超过阈值转入死信队列。
重试次数监控 记录消息重试次数和原因便于分析系统瓶颈。
隔离重试流量 使用独立队列和消费者组处理重试消息避免影响正常流量。
熔断机制 若某类消息持续失败如依赖服务宕机暂时停止消费并触发告警。
自动化测试 模拟消息重发场景验证幂等性和补偿逻辑的正确性。
五、总结【重点看总结】
消息重发与补充策略的核心目标是 平衡可靠性与系统复杂性
自动重试解决临时性故障如网络抖动。死信队列兜底无法自动修复的异常。补偿机制最终确保业务状态一致。幂等性防御重试导致的重复操作。 待完善
