消息队列的作用(消息队列的作用知乎)

本文目录：

• 1、消息队列(MQ)消息延迟及过滤设计方案
• 2、消息队列之消息总线的运用
• 3、分布式缓存主要用在高并发环境下的作用？
• 4、Redis的主要功能？
• 5、Redis 实现 消息队列

消息队列(MQ)消息延迟及过滤设计方案

消息队列接收到请求后，会将消息顺序写入Physic log文件，对于延迟消息，将消息按照将要投递的时间，以小时为单位异步分割存储，每个小时的消息顺序写入延迟物理文件delay log，并把索引存储在delay index文件，索引记录消息在delay log中的{offset，size，投递时间戳}元信息，基于有限内存以及延迟消息分发特性，我们仅将最近两个小时的delay log文件序列采用mmap内存映射机制进行读写，延迟2个小时以上的消息直接写入磁盘文件。

但消息分发时，每次需要将一个小时的的索引文件，全部加载到内存，由于每个小时的消息索引是顺序写入delay index的，而消息分发投递时间又是随机的，写入顺序与消息投递顺序并不一致，所以，索引加载到内存后，需要按照消息具体投递的秒级时间戳进行排序，再根据排序后的索引读取delay log中的消息进行分发投递。

这种存储方案有以下问题： 1.一次需要加载整个小时的消息索引到内存，若并发比较高，内存压力比较大。 2.按照消息投递秒级时间戳进行排序后，实时到来的新的消息，需要实时插入排序，性能较低，延迟大。

为了解决上述问题，我们将delay index中索引元信息{offset, size, 投递时间戳}改为{offset, size, localIndex, globalIndex, preGlobalIndex }，其中：

通过globalIndex可以直接定位到delay index中的索引单元，从而确定delay log中的一条消息，而preGlobalIndex又可以定位到同一秒内的上一条消息，因此只要落地存储每个小时，每秒最后一条消息的索引ID，即可逆序查出每秒所有消息。一个小时内只有3600秒，只需要将3600个16字节的索引ID加载到内存，即可实现每秒消息的实时加载。

为了降低消息分发延迟，可将最近10s的消息索引提前预加载到内存，对于实时接收到的消息，根据时间戳匹配到对应的秒，更新这一秒最新一条消息的索引globalIndex与逆向索引preGlobalIndex，不需要做排序，消息插入与读取的复杂度都为O(1)。

采用的由数组加链表实现的多级时间轮机制，分别是秒级和小时级，小时级时间轮前移一个槽，对应秒级时间轮旋转一圈，秒级时间轮上一共3600个槽，每个槽的时间跨度最大为1s，时间轮每秒前移一个槽。小时级实践论每个槽时间跨度是1小时，每小时移动一个槽，将后面两个小时的delay log开启内存映射，同时清除两个小时之前delay log文件内存映射。

当我们只有一个2小时5分钟的消息发送时，秒级时间轮需要推动2圈后即小时级时间轮移动2个槽，剩5分钟的延迟，再降级到秒级时间轮。这叫造成了时间轮的空转。

一般会把每个使用到的槽都会放到DelayQueue中，然后根据DelayQueue来 协助时间轮的推进 ，防止空推进的情况。例如，当有延迟500s的任务时，除了挂载到时间轮外，我们还会把其放到DelayQueue中，这样DelayQueue的头结点为延迟500s，如果期间没有小于500s的延迟任务再加进来时，我们只需要等待500s，时间轮推进一次即可。如果有小于500s的定时任务新加进来，我们只需要唤醒DelayQueue，重新计算等待时间即可。

即当有定时任务新增时，如果对应槽为新槽（即新增任务为该槽的第一个任务），在DelayQueue中增加延迟任务，并判断是否为头结点，是的话唤醒DelayQueue重新计算等待时间。

当master发生漂移或者网络异常时，时间轮分发控制需要从原master节点切换到新的master节点。为了保证分发状态的连续性与一致性，master节点定时每隔50ms分别将两个时间轮上分发的tick信息同步到其它slave节点。通过tick可确定具体分发到第几秒，但不能确定分发到这一秒的第几条消息，为此二级时间轮增加同步了一个参数localIndex，记录当前秒分发到第几条消息，并且每个节点都会定时将分发状态持久化。

每当master发生切换时，原master节点切换为slave，会立即停止当前时间轮的分发任务，并清空分发状态；而新的master节点根据当前已同步过来的分发状态初始化两级时间轮，但master切换会有一定的延迟周期或者极端情况下不同节点间时钟存在偏差，新的master初始完时间轮的tick后，该tick对应的秒级时间戳有可能与节点实际时间不一致，启动分发任务前需要做特殊调整，若tick时间戳小于当前时间，则分发任务sleep等待直至时间对齐，若tick时间戳大于当前时间，说明存在已到期的消息未分发，此时连续推进tick迁移，并对到期消息直接异步投递，直到tick对应时间戳小于当前时间。

正常master切换分为两种情况，一种为主动释放master，如节点重启与master负载均衡过程，这种情况节点在drop master之前，会首先同步时间轮分发状态到其它slave节点，此时master切换时间轮分发时完全连续一致的；另一种是一些异常情况下master被动漂移，此时新的master节点上时间轮分发状态可能存在最大50ms的延迟，会出现部分消息重复分发现象。把时间轮分发状态信息封装到到期投递消息协议扩展字段中，paxos请求同步消息时携带时间轮状态，即可做到实时同步。

众所周知，RocketMQ是支持消息过滤的，即发送消息时，可以给消息设置一个TAG。订阅主题的时候，可以设置只消费携带某些TAG的消息，起到消息过滤的作用。

客户端拉取消息时，在服务端得到tag的hash集合codeSet，然后从ConsumerQueue获取一条记录，判断记录的hashCode是否在codeSet中，以达到消息过滤的目的，决定是否将该消息发送给consumer。

因为Hash存在冲突，过滤不完全准确，所以，客户端收到消息后，会进行再次精准过滤。

还有一种过滤方式，把TAG通过哈希转换为long，索引中保存所有TAG的哈希值按位或的结果。当拉取消息时，通过订阅设置的TAG哈希值与索引中的哈希值进行按位与操作，如果结果等于订阅设置的TAG哈希值，说明该索引对应的消息可能符合条件，二次精准过滤依旧在客户端来做；否则，一定不符合条件，直接过滤掉。

消息队列之消息总线的运用

鉴于自己公司的架构中看到消息总线的运用，记录一下MQ的解耦作用

A 系统发送数据到 BCD 三个系统，通过接口调用发送。如果 E 系统也要这个数据呢？那如果 C 系统现在不需要了呢？

在这个场景中，A 系统跟其它系统耦合，A 系统产生一条数据，很多系统都需要从 A 系统获取数据。那么A 系统要考虑 其他四个系统如果宕机的情况。

如果使用 MQ，A 系统产生一条数据，发送到 MQ 里面去，哪个系统需要数据，就自己去 MQ 里面消费。如果新系统需要数据，直接从 MQ 里消费即可；如果某个系统不需要这条数据了，就取消对 MQ 消息的消费即可。这样下来，A 系统不需要去考虑要给谁发送数据，也不需要考虑人家是否调用成功、失败超时等情况。

公司架构

微服务治理的消息总线就是考虑一下系统中是否有一个系统或者一个模块，调用了多个系统或者模块，互相之间的调用很复杂，维护起来很麻烦。但是其实这个调用是不需要直接同步调用接口的，如果是这样的场景，可以用 MQ 给它异步化解耦。

分布式缓存主要用在高并发环境下的作用？

分布式缓存主要用在高并发环境下，减轻数据库的压力，提高系统的响应速度和并发吞吐。当大量的读、写请求涌向数据库时，磁盘的处理速度与内存显然不在一个量级，因此，在数据库之前加一层缓存，能够显著提高系统的响应速度，并降低数据库的压力。作为传统的关系型数据库，MySQL提供完整的ACID操作，支持丰富的数据类型、强大的关联查询、where语句等，能够非常客易地建立查询索引，执行复杂的内连接、外连接、求和、排序、分组等操作，并且支持存储过程、函数等功能，产品成熟度高，功能强大。但是，对于需要应对高并发访问并且存储海量数据的场景来说，出于对性能的考虑，不得不放弃很多传统关系型数据库原本强大的功能，牺牲了系统的易用性，并且使得系统的设计和管理变得更为复杂。这也使得在过去几年中，流行着另一种新的存储解决方案——NoSQL，它与传统的关系型数据库最大的差别在于，它不使用SQL作为查询语言来查找数据，而采用key-value形式进行查找，提供了更高的查询效率及吞吐，并且能够更加方便地进行扩展，存储海量数据，在数千个节点上进行分区，自动进行数据的复制和备份。在分布式系统中，消息作为应用间通信的一种方式，得到了十分广泛的应用。消息可以被保存在队列中，直到被接收者取出，由于消息发送者不需要同步等待消息接收者的响应，消息的异步接收降低了系统集成的耦合度，提升了分布式系统协作的效率，使得系统能够更快地响应用户，提供更高的吞吐。

当系统处于峰值压力时，分布式消息队列还能够作为缓冲，削峰填谷，缓解集群的压力，避免整个系统被压垮。垂直化的搜索引擎在分布式系统中是一个非常重要的角色，它既能够满足用户对于全文检索、模糊匹配的需求，解决数据库like查询效率低下的问题，又能够解决分布式环境下，由于采用分库分表，或者使用NoSQL数据库，导致无法进行多表关联或者进行复杂查询的问题。

Redis的主要功能？

缓存：这应该是 Redis 最主要的功能了，也是大型网站必备机制，合理地使用缓存不仅可以加 快数据的访问速度，而且能够有效地降低后端数据源的压力。

共享Session：对于一些依赖 session 功能的服务来说，如果需要从单机变成集群的话，可以选择 redis 来统一管理 session。消息队列系统：消息队列系统可以说是一个大型网站的必备基础组件，因为其具有业务 解耦、非实时业务削峰等特性。Redis提供了发布订阅功能和阻塞队列的功 能，虽然和专业的消息队列比还不够足够强大，但是对于一般的消息队列功能基本可以满足。比如在分布式爬虫系统中，使用 redis 来统一管理 url队列。

分布式锁：在分布式服务中。可以利用Redis的setnx功能来编写分布式的锁，虽然这个可能不是太常用。 当然还有诸如排行榜、点赞功能都可以使用 Redis 来实现，但是 Redis 也不是什么都可以做，比如数据量特别大时，不适合 Redis，我们知道 Redis 是基于内存的，虽然内存很便宜，但是如果你每天的数据量特别大，比如几亿条的用户行为日志数据，用 Redis 来存储的话，成本相当的高。

Redis 实现 消息队列

虽然消费者是异步处理消息，但是，消费者仍然需要按照生产者发送消息的顺序来处理消息，避免后发送的消息被先处理了。对于要求消息保序的场景来说，一旦出现这种消息被乱序处理的情况，就可能会导致业务逻辑被错误执行，从而给业务方造成损失。

消费者从消息队列读取消息时，有时会因为网络堵塞而出现消息重传的情况。此时，消费者可能会收到多条重复的消息。对于重复的消息，消费者如果多次处理的话，就可能造成一个业务逻辑被多次执行，如果业务逻辑正好是要修改数据，那就会出现数据被多次修改的问题了。

消费者在处理消息的时候，还可能出现因为故障或宕机导致消息没有处理完成的情况。此时，消息队列需要能提供消息可靠性的保证，也就是说，当消费者重启后，可以重新读取消息再次进行处理，否则，就会出现消息漏处理的问题了。

List 本身就是按先进先出的顺序对数据进行存取的，所以，如果使用 List 作为消息队列保存消息的话，就已经能满足消息保序的需求了。

生产者提供一个消息id，消费者要把已经处理过的消息 ID 号记录下来。当收到一条消息后，消费者程序就可以对比收到的消息 ID 和记录的已处理过的消息 ID，来判断当前收到的消息有没有经过处理。如果已经处理过，那么，消费者程序就不再进行处理了。

为了留存消息，List 类型提供了 BRPOPLPUSH 命令，这个命令的作用是让消费者程序从一个 List 中读取消息，同时，Redis 会把这个消息再插入到另一个 List（可以叫作备份 List）留存。这样一来，如果消费者程序读了消息但没能正常处理，等它重启后，就可以从备份 List 中重新读取消息并进行处理了。

在生产者往 List 中写入数据时，List 并不会主动地通知消费者有新消息写入，如果消费者想要及时处理消息，就需要在程序中不停地调用 RPOP 命令（比如使用一个 while(1) 循环）。如果有新消息写入，RPOP 命令就会返回结果，否则，RPOP 命令返回空值，再继续循环。所以，即使没有新消息写入 List，消费者也要不停地调用 RPOP 命令，这就会导致消费者程序的 CPU 一直消耗在执行 RPOP 命令上，带来不必要的性能损失。为了解决这个问题，Redis 提供了 BRPOP 命令。BRPOP 命令也称为阻塞式读取，客户端在没有读到队列数据时，自动阻塞，直到有新的数据写入队列，再开始读取新数据。和消费者程序自己不停地调用 RPOP 命令相比，这种方式能节省 CPU 开销。

发送消息日志

消费消息日志

Streams 是 Redis 专门为消息队列设计的数据类型，它提供了丰富的消息队列操作命令。

XADD：插入消息，保证有序，可以自动生成全局唯一 ID；

XREAD：用于读取消息，可以按 ID 读取数据；

XREADGROUP：按消费组形式读取消息；

XPENDING ：XPENDING 命令可以用来查询每个消费组内所有消费者已读取但尚未确认的消息，

XACK：XACK 命令用于向消息队列确认消息处理已完成。

监听器

监听器容器

初始化消息队列(integral-queue)和消费者组(integral-group),执行RedisMqTest.initMessageQueue()即可

可参考List实现的处理方式

可使用RedisMqTest.getPendingQueue()获取未成功消费的消息队列进行处理

本文来源：https://www.yuntue.com/post/116534.html | 云服务器网，转载请注明出处！