长期的应试教育让我花了太久才明白：学习其实是一件只与自己相关的事。 现在的我，更愿意主动探索那些真正能激发好奇心、让生活变得更有趣的事物。 如果你有能够启发我、让我产生兴趣的想法、机会或故事，欢迎写信给我： （ljr0115@outlook.com） 我会优先回复那些 与我的时间安排契合、或与我当下关注方向相吻合 的邮件。
AI 摘要 文章系统梳理RocketMQ核心知识点：从消息队列诞生的异步、解耦、削峰三大价值，到RocketMQ主题模型、NameServer+Broker架构、零拷贝高性能存储、同步/异步刷盘与复制、事务消息、顺序与重复消费、消息堆积等常见问题及解决方案，并给出代码级最佳实践。 此摘要根据文章内容自动生成，仅用于文章内容的解释与总结title: RocketMQ常见问题总结category: 高性能tag:RocketMQ消息队列消息队列扫盲消息队列顾名思义就是存放消息的队列，队列我就不解释了，别告诉我你连队列都不知道是啥吧？所以问题并不是消息队列是什么，而是 消息队列为什么会出现？消息队列能用来干什么？用它来干这些事会带来什么好处？消息队列会带来副作用吗？消息队列为什么会出现？消息队``列算是作为后端程序员的一个必备技能吧，因为分布式应用必定涉及到各个系统之间的通信问题，这个时候消息队列也应运而生了。可以说分布式的产生是消息队列的基础，而分布式怕是一个很古老的概念了吧，所以消息队列也是一个很古老的中间件了。消息队列能用来干什么？异步你可能会反驳我，应用之间的通信又不是只能由消息队列解决，好好的通信为什么中间非要插一个消息队列呢？我不能直接进行通信吗？很好 👍，你又提出了一个概念，同步通信。就比如现在业界使用比较多的 Dubbo 就是一个适用于各个系统之间同步通信的 RPC 框架。我来举个 🌰 吧，比如我们有一个购票系统，需求是用户在购买完之后能接收到购买完成的短信。我们省略中间的网络通信时间消耗，假如购票系统处理需要 150ms ，短信系统处理需要 200ms ，那么整个处理流程的时间消耗就是 150ms + 200ms = 350ms。当然，乍看没什么问题。可是仔细一想你就感觉有点问题，我用户购票在购票系统的时候其实就已经完成了购买，而我现在通过同步调用非要让整个请求拉长时间，而短信系统这玩意又不是很有必要，它仅仅是一个辅助功能增强用户体验感而已。我现在整个调用流程就有点 头重脚轻 的感觉了，购票是一个不太耗时的流程，而我现在因为同步调用，非要等待发送短信这个比较耗时的操作才返回结果。那我如果再加一个发送邮件呢？这样整个系统的调用链又变长了，整个时间就变成了 550ms。当我们在学生时代需要在食堂排队的时候，我们和食堂大妈就是一个同步的模型。我们需要告诉食堂大妈：“姐姐，给我加个鸡腿，再加个酸辣土豆丝，帮我浇点汁上去，多打点饭哦 😋😋😋” 咦~~~ 为了多吃点，真恶心。然后大妈帮我们打饭配菜，我们看着大妈那颤抖的手和掉落的土豆丝不禁咽了咽口水。最终我们从大妈手中接过饭菜然后去寻找座位了...回想一下，我们在给大妈发送需要的信息之后我们是 同步等待大妈给我配好饭菜 的，上面我们只是加了鸡腿和土豆丝，万一我再加一个番茄牛腩，韭菜鸡蛋，这样是不是大妈打饭配菜的流程就会变长，我们等待的时间也会相应的变长。那后来，我们工作赚钱了有钱去饭店吃饭了，我们告诉服务员来一碗牛肉面加个荷包蛋 (传达一个消息) ，然后我们就可以在饭桌上安心的玩手机了 (干自己其他事情) ，等到我们的牛肉面上了我们就可以吃了。这其中我们也就传达了一个消息，然后我们又转过头干其他事情了。这其中虽然做面的时间没有变短，但是我们只需要传达一个消息就可以干其他事情了，这是一个 异步 的概念。所以，为了解决这一个问题，聪明的程序员在中间也加了个类似于服务员的中间件——消息队列。这个时候我们就可以把模型给改造了。这样，我们在将消息存入消息队列之后我们就可以直接返回了(我们告诉服务员我们要吃什么然后玩手机)，所以整个耗时只是 150ms + 10ms = 160ms。但是你需要注意的是，整个流程的时长是没变的，就像你仅仅告诉服务员要吃什么是不会影响到做面的速度的。解耦回到最初同步调用的过程，我们写个伪代码简单概括一下。那么第二步，我们又添加了一个发送邮件，我们就得重新去修改代码，如果我们又加一个需求：用户购买完还需要给他加积分，这个时候我们是不是又得改代码？如果你觉得还行，那么我这个时候不要发邮件这个服务了呢，我是不是又得改代码，又得重启应用？这样改来改去是不是很麻烦，那么 此时我们就用一个消息队列在中间进行解耦 。你需要注意的是，我们后面的发送短信、发送邮件、添加积分等一些操作都依赖于上面的 result ，这东西抽象出来就是购票的处理结果呀，比如订单号，用户账号等等，也就是说我们后面的一系列服务都是需要同样的消息来进行处理。既然这样，我们是不是可以通过 “广播消息” 来实现。我上面所讲的“广播”并不是真正的广播，而是接下来的系统作为消费者去 订阅 特定的主题。比如我们这里的主题就可以叫做 订票 ，我们购买系统作为一个生产者去生产这条消息放入消息队列，然后消费者订阅了这个主题，会从消息队列中拉取消息并消费。就比如我们刚刚画的那张图，你会发现，在生产者这边我们只需要关注 生产消息到指定主题中 ，而 消费者只需要关注从指定主题中拉取消息 就行了。如果没有消息队列，每当一个新的业务接入，我们都要在主系统调用新接口、或者当我们取消某些业务，我们也得在主系统删除某些接口调用。有了消息队列，我们只需要关心消息是否送达了队列，至于谁希望订阅，接下来收到消息如何处理，是下游的事情，无疑极大地减少了开发和联调的工作量。削峰我们再次回到一开始我们使用同步调用系统的情况，并且思考一下，如果此时有大量用户请求购票整个系统会变成什么样？如果，此时有一万的请求进入购票系统，我们知道运行我们主业务的服务器配置一般会比较好，所以这里我们假设购票系统能承受这一万的用户请求，那么也就意味着我们同时也会出现一万调用发短信服务的请求。而对于短信系统来说并不是我们的主要业务，所以我们配备的硬件资源并不会太高，那么你觉得现在这个短信系统能承受这一万的峰值么，且不说能不能承受，系统会不会 直接崩溃 了？短信业务又不是我们的主业务，我们能不能 折中处理 呢？如果我们把购买完成的信息发送到消息队列中，而短信系统 尽自己所能地去消息队列中取消息和消费消息 ，即使处理速度慢一点也无所谓，只要我们的系统没有崩溃就行了。留得江山在，还怕没柴烧？你敢说每次发送验证码的时候是一发你就收到了的么？消息队列能带来什么好处？其实上面我已经说了。异步、解耦、削峰。 哪怕你上面的都没看懂也千万要记住这六个字，因为他不仅是消息队列的精华，更是编程和架构的精华。消息队列会带来副作用吗？没有哪一门技术是“银弹”，消息队列也有它的副作用。比如，本来好好的两个系统之间的调用，我中间加了个消息队列，如果消息队列挂了怎么办呢？是不是 降低了系统的可用性 ？那这样是不是要保证 HA(高可用)？是不是要搞集群？那么我 整个系统的复杂度是不是上升了 ？抛开上面的问题不讲，万一我发送方发送失败了，然后执行重试，这样就可能产生重复的消息。或者我消费端处理失败了，请求重发，这样也会产生重复的消息。对于一些微服务来说，消费重复消息会带来更大的麻烦，比如增加积分，这个时候我加了多次是不是对其他用户不公平？那么，又 如何解决重复消费消息的问题 呢？如果我们此时的消息需要保证严格的顺序性怎么办呢？比如生产者生产了一系列的有序消息(对一个 id 为 1 的记录进行删除增加修改)，但是我们知道在发布订阅模型中，对于主题是无顺序的，那么这个时候就会导致对于消费者消费消息的时候没有按照生产者的发送顺序消费，比如这个时候我们消费的顺序为修改删除增加，如果该记录涉及到金额的话是不是会出大事情？那么，又 如何解决消息的顺序消费问题 呢？就拿我们上面所讲的分布式系统来说，用户购票完成之后是不是需要增加账户积分？在同一个系统中我们一般会使用事务来进行解决，如果用 Spring 的话我们在上面伪代码中加入 @Transactional 注解就好了。但是在不同系统中如何保证事务呢？总不能这个系统我扣钱成功了你那积分系统积分没加吧？或者说我这扣钱明明失败了，你那积分系统给我加了积分。那么，又如何 解决分布式事务问题 呢？我们刚刚说了，消息队列可以进行削峰操作，那如果我的消费者如果消费很慢或者生产者生产消息很快，这样是不是会将消息堆积在消息队列中？那么，又如何 解决消息堆积的问题 呢？可用性降低，复杂度上升，又带来一系列的重复消费，顺序消费，分布式事务，消息堆积的问题，这消息队列还怎么用啊 😵？别急，办法总是有的。RocketMQ 是什么？哇，你个混蛋！上面给我抛出那么多问题，你现在又讲 RocketMQ ，还让不让人活了？！🤬别急别急，话说你现在清楚 MQ 的构造吗，我还没讲呢，我们先搞明白 MQ 的内部构造，再来看看如何解决上面的一系列问题吧，不过你最好带着问题去阅读和了解喔。RocketMQ 是一个 队列模型 的消息中间件，具有高性能、高可靠、高实时、分布式 的特点。它是一个采用 Java 语言开发的分布式的消息系统，由阿里巴巴团队开发，在 2016 年底贡献给 Apache，成为了 Apache 的一个顶级项目。 在阿里内部，RocketMQ 很好地服务了集团大大小小上千个应用，在每年的双十一当天，更有不可思议的万亿级消息通过 RocketMQ 流转。废话不多说，想要了解 RocketMQ 历史的同学可以自己去搜寻资料。听完上面的介绍，你只要知道 RocketMQ 很快、很牛、而且经历过双十一的实践就行了！队列模型和主题模型是什么？在谈 RocketMQ 的技术架构之前，我们先来了解一下两个名词概念——队列模型 和 主题模型 。首先我问一个问题，消息队列为什么要叫消息队列？你可能觉得很弱智，这玩意不就是存放消息的队列嘛？不叫消息队列叫什么？的确，早期的消息中间件是通过 队列 这一模型来实现的，可能是历史原因，我们都习惯把消息中间件成为消息队列。但是，如今例如 RocketMQ、Kafka 这些优秀的消息中间件不仅仅是通过一个 队列 来实现消息存储的。队列模型就像我们理解队列一样，消息中间件的队列模型就真的只是一个队列。。。我画一张图给大家理解。在一开始我跟你提到了一个 “广播” 的概念，也就是说如果我们此时我们需要将一个消息发送给多个消费者(比如此时我需要将信息发送给短信系统和邮件系统)，这个时候单个队列即不能满足需求了。当然你可以让 Producer 生产消息放入多个队列中，然后每个队列去对应每一个消费者。问题是可以解决，创建多个队列并且复制多份消息是会很影响资源和性能的。而且，这样子就会导致生产者需要知道具体消费者个数然后去复制对应数量的消息队列，这就违背我们消息中间件的 解耦 这一原则。主题模型那么有没有好的方法去解决这一个问题呢？有，那就是 主题模型 或者可以称为 发布订阅模型 。感兴趣的同学可以去了解一下设计模式里面的观察者模式并且手动实现一下，我相信你会有所收获的。在主题模型中，消息的生产者称为 发布者(Publisher) ，消息的消费者称为 订阅者(Subscriber) ，存放消息的容器称为 主题(Topic) 。其中，发布者将消息发送到指定主题中，订阅者需要 提前订阅主题 才能接受特定主题的消息。RocketMQ 中的消息模型RocketMQ 中的消息模型就是按照 主题模型 所实现的。你可能会好奇这个 主题 到底是怎么实现的呢？你上面也没有讲到呀！其实对于主题模型的实现来说每个消息中间件的底层设计都是不一样的，就比如 Kafka 中的 分区 ，RocketMQ 中的 队列 ，RabbitMQ 中的 Exchange 。我们可以理解为 主题模型/发布订阅模型 就是一个标准，那些中间件只不过照着这个标准去实现而已。所以，RocketMQ 中的 主题模型 到底是如何实现的呢？首先我画一张图，大家尝试着去理解一下。我们可以看到在整个图中有 Producer Group、Topic、Consumer Group 三个角色，我来分别介绍一下他们。Producer Group 生产者组：代表某一类的生产者，比如我们有多个秒杀系统作为生产者，这多个合在一起就是一个 Producer Group 生产者组，它们一般生产相同的消息。Consumer Group 消费者组：代表某一类的消费者，比如我们有多个短信系统作为消费者，这多个合在一起就是一个 Consumer Group 消费者组，它们一般消费相同的消息。Topic 主题：代表一类消息，比如订单消息，物流消息等等。你可以看到图中生产者组中的生产者会向主题发送消息，而 主题中存在多个队列，生产者每次生产消息之后是指定主题中的某个队列发送消息的。每个主题中都有多个队列(分布在不同的 Broker中，如果是集群的话，Broker又分布在不同的服务器中)，集群消费模式下，一个消费者集群多台机器共同消费一个 topic 的多个队列，一个队列只会被一个消费者消费。如果某个消费者挂掉，分组内其它消费者会接替挂掉的消费者继续消费。就像上图中 Consumer1 和 Consumer2 分别对应着两个队列，而 Consumer3 是没有队列对应的，所以一般来讲要控制 消费者组中的消费者个数和主题中队列个数相同 。当然也可以消费者个数小于队列个数，只不过不太建议。如下图。每个消费组在每个队列上维护一个消费位置 ，为什么呢？因为我们刚刚画的仅仅是一个消费者组，我们知道在发布订阅模式中一般会涉及到多个消费者组，而每个消费者组在每个队列中的消费位置都是不同的。如果此时有多个消费者组，那么消息被一个消费者组消费完之后是不会删除的(因为其它消费者组也需要呀)，它仅仅是为每个消费者组维护一个 消费位移(offset) ，每次消费者组消费完会返回一个成功的响应，然后队列再把维护的消费位移加一，这样就不会出现刚刚消费过的消息再一次被消费了。可能你还有一个问题，为什么一个主题中需要维护多个队列 ？答案是 提高并发能力 。的确，每个主题中只存在一个队列也是可行的。你想一下，如果每个主题中只存在一个队列，这个队列中也维护着每个消费者组的消费位置，这样也可以做到 发布订阅模式 。如下图。但是，这样我生产者是不是只能向一个队列发送消息？又因为需要维护消费位置所以一个队列只能对应一个消费者组中的消费者，这样是不是其他的 Consumer 就没有用武之地了？从这两个角度来讲，并发度一下子就小了很多。所以总结来说，RocketMQ 通过使用在一个 Topic 中配置多个队列并且每个队列维护每个消费者组的消费位置 实现了 主题模式/发布订阅模式 。RocketMQ 的架构图讲完了消息模型，我们理解起 RocketMQ 的技术架构起来就容易多了。RocketMQ 技术架构中有四大角色 NameServer、Broker、Producer、Consumer 。我来向大家分别解释一下这四个角色是干啥的。Broker：主要负责消息的存储、投递和查询以及服务高可用保证。说白了就是消息队列服务器嘛，生产者生产消息到 Broker ，消费者从 Broker 拉取消息并消费。这里，我还得普及一下关于 Broker、Topic 和 队列的关系。上面我讲解了 Topic 和队列的关系——一个 Topic 中存在多个队列，那么这个 Topic 和队列存放在哪呢？一个 Topic 分布在多个 Broker上，一个 Broker 可以配置多个 Topic ，它们是多对多的关系。如果某个 Topic 消息量很大，应该给它多配置几个队列(上文中提到了提高并发能力)，并且 尽量多分布在不同 Broker 上，以减轻某个 Broker 的压力 。Topic 消息量都比较均匀的情况下，如果某个 broker 上的队列越多，则该 broker 压力越大。所以说我们需要配置多个 Broker。NameServer：不知道你们有没有接触过 ZooKeeper 和 Spring Cloud 中的 Eureka ，它其实也是一个 注册中心 ，主要提供两个功能：Broker 管理 和 路由信息管理 。说白了就是 Broker 会将自己的信息注册到 NameServer 中，此时 NameServer 就存放了很多 Broker 的信息(Broker 的路由表)，消费者和生产者就从 NameServer 中获取路由表然后照着路由表的信息和对应的 Broker 进行通信(生产者和消费者定期会向 NameServer 去查询相关的 Broker 的信息)。Producer：消息发布的角色，支持分布式集群方式部署。说白了就是生产者。Consumer：消息消费的角色，支持分布式集群方式部署。支持以 push 推，pull 拉两种模式对消息进行消费。同时也支持集群方式和广播方式的消费，它提供实时消息订阅机制。说白了就是消费者。听完了上面的解释你可能会觉得，这玩意好简单。不就是这样的么？嗯？你可能会发现一个问题，这老家伙 NameServer 干啥用的，这不多余吗？直接 Producer、Consumer 和 Broker 直接进行生产消息，消费消息不就好了么？但是，我们上文提到过 Broker 是需要保证高可用的，如果整个系统仅仅靠着一个 Broker 来维持的话，那么这个 Broker 的压力会不会很大？所以我们需要使用多个 Broker 来保证 负载均衡 。如果说，我们的消费者和生产者直接和多个 Broker 相连，那么当 Broker 修改的时候必定会牵连着每个生产者和消费者，这样就会产生耦合问题，而 NameServer 注册中心就是用来解决这个问题的。如果还不是很理解的话，可以去看我介绍 Spring Cloud 的那篇文章，其中介绍了 Eureka 注册中心。当然，RocketMQ 中的技术架构肯定不止前面那么简单，因为上面图中的四个角色都是需要做集群的。我给出一张官网的架构图，大家尝试理解一下。其实和我们最开始画的那张乞丐版的架构图也没什么区别，主要是一些细节上的差别。听我细细道来 🤨。第一、我们的 Broker 做了集群并且还进行了主从部署 ，由于消息分布在各个 Broker 上，一旦某个 Broker 宕机，则该Broker 上的消息读写都会受到影响。所以 Rocketmq 提供了 master/slave 的结构，salve 定时从 master 同步数据(同步刷盘或者异步刷盘)，如果 master 宕机，则 slave 提供消费服务，但是不能写入消息 (后面我还会提到哦)。第二、为了保证 HA ，我们的 NameServer 也做了集群部署，但是请注意它是 去中心化 的。也就意味着它没有主节点，你可以很明显地看出 NameServer 的所有节点是没有进行 Info Replicate 的，在 RocketMQ 中是通过 单个 Broker 和所有 NameServer 保持长连接 ，并且在每隔 30 秒 Broker 会向所有 Nameserver 发送心跳，心跳包含了自身的 Topic 配置信息，这个步骤就对应这上面的 Routing Info 。第三、在生产者需要向 Broker 发送消息的时候，需要先从 NameServer 获取关于 Broker 的路由信息，然后通过 轮询 的方法去向每个队列中生产数据以达到 负载均衡 的效果。第四、消费者通过 NameServer 获取所有 Broker 的路由信息后，向 Broker 发送 Pull 请求来获取消息数据。Consumer 可以以两种模式启动—— 广播（Broadcast）和集群（Cluster）。广播模式下，一条消息会发送给 同一个消费组中的所有消费者 ，集群模式下消息只会发送给一个消费者。RocketMQ 功能特性消息普通消息普通消息一般应用于微服务解耦、事件驱动、数据集成等场景，这些场景大多数要求数据传输通道具有可靠传输的能力，且对消息的处理时机、处理顺序没有特别要求。以在线的电商交易场景为例，上游订单系统将用户下单支付这一业务事件封装成独立的普通消息并发送至 RocketMQ 服务端，下游按需从服务端订阅消息并按照本地消费逻辑处理下游任务。每个消息之间都是相互独立的，且不需要产生关联。另外还有日志系统，以离线的日志收集场景为例，通过埋点组件收集前端应用的相关操作日志，并转发到 RocketMQ 。普通消息生命周期初始化：消息被生产者构建并完成初始化，待发送到服务端的状态。待消费：消息被发送到服务端，对消费者可见，等待消费者消费的状态。消费中：消息被消费者获取，并按照消费者本地的业务逻辑进行处理的过程。 此时服务端会等待消费者完成消费并提交消费结果，如果一定时间后没有收到消费者的响应，RocketMQ 会对消息进行重试处理。消费提交：消费者完成消费处理，并向服务端提交消费结果，服务端标记当前消息已经被处理（包括消费成功和失败）。RocketMQ 默认支持保留所有消息，此时消息数据并不会立即被删除，只是逻辑标记已消费。消息在保存时间到期或存储空间不足被删除前，消费者仍然可以回溯消息重新消费。消息删除：RocketMQ 按照消息保存机制滚动清理最早的消息数据，将消息从物理文件中删除。定时消息在分布式定时调度触发、任务超时处理等场景，需要实现精准、可靠的定时事件触发。使用 RocketMQ 的定时消息可以简化定时调度任务的开发逻辑，实现高性能、可扩展、高可靠的定时触发能力。定时消息仅支持在 MessageType 为 Delay 的主题内使用，即定时消息只能发送至类型为定时消息的主题中，发送的消息的类型必须和主题的类型一致。在 4.x 版本中，只支持延时消息，默认分为 18 个等级分别为：1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h，也可以在配置文件中增加自定义的延时等级和时长。在 5.x 版本中，开始支持定时消息，在构造消息时提供了 3 个 API 来指定延迟时间或定时时间。基于定时消息的超时任务处理具备如下优势：精度高、开发门槛低：基于消息通知方式不存在定时阶梯间隔。可以轻松实现任意精度事件触发，无需业务去重。高性能可扩展：传统的数据库扫描方式较为复杂，需要频繁调用接口扫描，容易产生性能瓶颈。RocketMQ 的定时消息具有高并发和水平扩展的能力。定时消息生命周期初始化：消息被生产者构建并完成初始化，待发送到服务端的状态。定时中：消息被发送到服务端，和普通消息不同的是，服务端不会直接构建消息索引，而是会将定时消息单独存储在定时存储系统中，等待定时时刻到达。待消费：定时时刻到达后，服务端将消息重新写入普通存储引擎，对下游消费者可见，等待消费者消费的状态。消费中：消息被消费者获取，并按照消费者本地的业务逻辑进行处理的过程。 此时服务端会等待消费者完成消费并提交消费结果，如果一定时间后没有收到消费者的响应，RocketMQ 会对消息进行重试处理。消费提交：消费者完成消费处理，并向服务端提交消费结果，服务端标记当前消息已经被处理（包括消费成功和失败）。RocketMQ 默认支持保留所有消息，此时消息数据并不会立即被删除，只是逻辑标记已消费。消息在保存时间到期或存储空间不足被删除前，消费者仍然可以回溯消息重新消费。消息删除：Apache RocketMQ 按照消息保存机制滚动清理最早的消息数据，将消息从物理文件中删除。定时消息的实现逻辑需要先经过定时存储等待触发，定时时间到达后才会被投递给消费者。因此，如果将大量定时消息的定时时间设置为同一时刻，则到达该时刻后会有大量消息同时需要被处理，会造成系统压力过大，导致消息分发延迟，影响定时精度。顺序消息顺序消息仅支持使用 MessageType 为 FIFO 的主题，即顺序消息只能发送至类型为顺序消息的主题中，发送的消息的类型必须和主题的类型一致。和普通消息发送相比，顺序消息发送必须要设置消息组。（推荐实现 MessageQueueSelector 的方式，见下文）。要保证消息的顺序性需要单一生产者串行发送。单线程使用 MessageListenerConcurrently 可以顺序消费，多线程环境下使用 MessageListenerOrderly 才能顺序消费。事务消息事务消息是 Apache RocketMQ 提供的一种高级消息类型，支持在分布式场景下保障消息生产和本地事务的最终一致性。简单来讲，就是将本地事务（数据库的 DML 操作）与发送消息合并在同一个事务中。例如，新增一个订单。在事务未提交之前，不发送订阅的消息。发送消息的动作随着事务的成功提交而发送，随着事务的回滚而取消。当然真正地处理过程不止这么简单，包含了半消息、事务监听和事务回查等概念，下面有更详细的说明。关于发送消息不建议单一进程创建大量生产者Apache RocketMQ 的生产者和主题是多对多的关系，支持同一个生产者向多个主题发送消息。对于生产者的创建和初始化，建议遵循够用即可、最大化复用原则，如果有需要发送消息到多个主题的场景，无需为每个主题都创建一个生产者。不建议频繁创建和销毁生产者Apache RocketMQ 的生产者是可以重复利用的底层资源，类似数据库的连接池。因此不需要在每次发送消息时动态创建生产者，且在发送结束后销毁生产者。这样频繁的创建销毁会在服务端产生大量短连接请求，严重影响系统性能。正确示例：Producer p = ProducerBuilder.build(); for (int i =0;i<n;i++){ Message m= MessageBuilder.build(); p.send(m); } p.shutdown();消费者分类PushConsumer高度封装的消费者类型，消费消息仅仅通过消费监听器监听并返回结果。消息的获取、消费状态提交以及消费重试都通过 RocketMQ 的客户端 SDK 完成。PushConsumer 的消费监听器执行结果分为以下三种情况：返回消费成功：以 Java SDK 为例，返回ConsumeResult.SUCCESS，表示该消息处理成功，服务端按照消费结果更新消费进度。返回消费失败：以 Java SDK 为例，返回ConsumeResult.FAILURE，表示该消息处理失败，需要根据消费重试逻辑判断是否进行重试消费。出现非预期失败：例如抛异常等行为，该结果按照消费失败处理，需要根据消费重试逻辑判断是否进行重试消费。具体实现可以参见这篇文章RocketMQ 对 pull 和 push 的实现。使用 PushConsumer 消费者消费时，不允许使用以下方式处理消息，否则 RocketMQ 无法保证消息的可靠性。错误方式一：消息还未处理完成，就提前返回消费成功结果。此时如果消息消费失败，RocketMQ 服务端是无法感知的，因此不会进行消费重试。错误方式二：在消费监听器内将消息再次分发到自定义的其他线程，消费监听器提前返回消费结果。此时如果消息消费失败，RocketMQ 服务端同样无法感知，因此也不会进行消费重试。PushConsumer 严格限制了消息同步处理及每条消息的处理超时时间，适用于以下场景：消息处理时间可预估：如果不确定消息处理耗时，经常有预期之外的长时间耗时的消息，PushConsumer 的可靠性保证会频繁触发消息重试机制造成大量重复消息。无异步化、高级定制场景：PushConsumer 限制了消费逻辑的线程模型，由客户端 SDK 内部按最大吞吐量触发消息处理。该模型开发逻辑简单，但是不允许使用异步化和自定义处理流程。SimpleConsumerSimpleConsumer 是一种接口原子型的消费者类型，消息的获取、消费状态提交以及消费重试都是通过消费者业务逻辑主动发起调用完成。一个来自官网的例子：// 消费示例：使用 SimpleConsumer 消费普通消息，主动获取消息处理并提交。 ClientServiceProvider provider = ClientServiceProvider.loadService(); String topic = "YourTopic"; FilterExpression filterExpression = new FilterExpression("YourFilterTag", FilterExpressionType.TAG); SimpleConsumer simpleConsumer = provider.newSimpleConsumerBuilder() // 设置消费者分组。 .setConsumerGroup("YourConsumerGroup") // 设置接入点。 .setClientConfiguration(ClientConfiguration.newBuilder().setEndpoints("YourEndpoint").build()) // 设置预绑定的订阅关系。 .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression)) // 设置从服务端接受消息的最大等待时间 .setAwaitDuration(Duration.ofSeconds(1)) .build(); try { // SimpleConsumer 需要主动获取消息，并处理。 List<MessageView> messageViewList = simpleConsumer.receive(10, Duration.ofSeconds(30)); messageViewList.forEach(messageView -> { System.out.println(messageView); // 消费处理完成后，需要主动调用 ACK 提交消费结果。 try { simpleConsumer.ack(messageView); } catch (ClientException e) { logger.error("Failed to ack message, messageId={}", messageView.getMessageId(), e); } }); } catch (ClientException e) { // 如果遇到系统流控等原因造成拉取失败，需要重新发起获取消息请求。 logger.error("Failed to receive message", e); }SimpleConsumer 适用于以下场景：消息处理时长不可控：如果消息处理时长无法预估，经常有长时间耗时的消息处理情况。建议使用 SimpleConsumer 消费类型，可以在消费时自定义消息的预估处理时长，若实际业务中预估的消息处理时长不符合预期，也可以通过接口提前修改。需要异步化、批量消费等高级定制场景：SimpleConsumer 在 SDK 内部没有复杂的线程封装，完全由业务逻辑自由定制，可以实现异步分发、批量消费等高级定制场景。需要自定义消费速率：SimpleConsumer 是由业务逻辑主动调用接口获取消息，因此可以自由调整获取消息的频率，自定义控制消费速率。PullConsumer施工中。。。消费者分组和生产者分组生产者分组RocketMQ 服务端 5.x 版本开始，生产者是匿名的，无需管理生产者分组（ProducerGroup）；对于历史版本服务端 3.x 和 4.x 版本，已经使用的生产者分组可以废弃无需再设置，且不会对当前业务产生影响。消费者分组消费者分组是多个消费行为一致的消费者的负载均衡分组。消费者分组不是具体实体而是一个逻辑资源。通过消费者分组实现消费性能的水平扩展以及高可用容灾。消费者分组中的订阅关系、投递顺序性、消费重试策略是一致的。订阅关系：Apache RocketMQ 以消费者分组的粒度管理订阅关系，实现订阅关系的管理和追溯。投递顺序性：Apache RocketMQ 的服务端将消息投递给消费者消费时，支持顺序投递和并发投递，投递方式在消费者分组中统一配置。消费重试策略： 消费者消费消息失败时的重试策略，包括重试次数、死信队列设置等。RocketMQ 服务端 5.x 版本：上述消费者的消费行为从关联的消费者分组中统一获取，因此，同一分组内所有消费者的消费行为必然是一致的，客户端无需关注。RocketMQ 服务端 3.x/4.x 历史版本：上述消费逻辑由消费者客户端接口定义，因此，您需要自己在消费者客户端设置时保证同一分组下的消费者的消费行为一致。(来自官方网站)如何解决顺序消费和重复消费？其实，这些东西都是我在介绍消息队列带来的一些副作用的时候提到的，也就是说，这些问题不仅仅挂钩于 RocketMQ ，而是应该每个消息中间件都需要去解决的。在上面我介绍 RocketMQ 的技术架构的时候我已经向你展示了 它是如何保证高可用的 ，这里不涉及运维方面的搭建，如果你感兴趣可以自己去官网上照着例子搭建属于你自己的 RocketMQ 集群。其实 Kafka 的架构基本和 RocketMQ 类似，只是它注册中心使用了 Zookeeper、它的 分区 就相当于 RocketMQ 中的 队列 。还有一些小细节不同会在后面提到。顺序消费在上面的技术架构介绍中，我们已经知道了 RocketMQ 在主题上是无序的、它只有在队列层面才是保证有序 的。这又扯到两个概念——普通顺序 和 严格顺序 。所谓普通顺序是指 消费者通过 同一个消费队列收到的消息是有顺序的 ，不同消息队列收到的消息则可能是无顺序的。普通顺序消息在 Broker 重启情况下不会保证消息顺序性 (短暂时间) 。所谓严格顺序是指 消费者收到的 所有消息 均是有顺序的。严格顺序消息 即使在异常情况下也会保证消息的顺序性 。但是，严格顺序看起来虽好，实现它可会付出巨大的代价。如果你使用严格顺序模式，Broker 集群中只要有一台机器不可用，则整个集群都不可用。你还用啥？现在主要场景也就在 binlog 同步。一般而言，我们的 MQ 都是能容忍短暂的乱序，所以推荐使用普通顺序模式。那么，我们现在使用了 普通顺序模式 ，我们从上面学习知道了在 Producer 生产消息的时候会进行轮询(取决你的负载均衡策略)来向同一主题的不同消息队列发送消息。那么如果此时我有几个消息分别是同一个订单的创建、支付、发货，在轮询的策略下这 三个消息会被发送到不同队列 ，因为在不同的队列此时就无法使用 RocketMQ 带来的队列有序特性来保证消息有序性了。那么，怎么解决呢？其实很简单，我们需要处理的仅仅是将同一语义下的消息放入同一个队列(比如这里是同一个订单)，那我们就可以使用 Hash 取模法 来保证同一个订单在同一个队列中就行了。RocketMQ 实现了两种队列选择算法，也可以自己实现轮询算法轮询算法就是向消息指定的 topic 所在队列中依次发送消息，保证消息均匀分布是 RocketMQ 默认队列选择算法最小投递延迟算法每次消息投递的时候统计消息投递的延迟，选择队列时优先选择消息延时小的队列，导致消息分布不均匀,按照如下设置即可。继承 MessageQueueSelector 实现SendResult sendResult = producer.send(msg, new MessageQueueSelector() { @Override public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) { //从mqs中选择一个队列,可以根据msg特点选择 return null; }特殊情况处理发送异常选择队列后会与 Broker 建立连接，通过网络请求将消息发送到 Broker 上，如果 Broker 挂了或者网络波动发送消息超时此时 RocketMQ 会进行重试。重新选择其他 Broker 中的消息队列进行发送，默认重试两次，可以手动设置。producer.setRetryTimesWhenSendFailed(5);消息过大消息超过 4k 时 RocketMQ 会将消息压缩后在发送到 Broker 上，减少网络资源的占用。重复消费emmm，就两个字—— 幂等 。在编程中一个幂等 操作的特点是其任意多次执行所产生的影响均与一次执行的影响相同。比如说，这个时候我们有一个订单的处理积分的系统，每当来一个消息的时候它就负责为创建这个订单的用户的积分加上相应的数值。可是有一次，消息队列发送给订单系统 FrancisQ 的订单信息，其要求是给 FrancisQ 的积分加上 500。但是积分系统在收到 FrancisQ 的订单信息处理完成之后返回给消息队列处理成功的信息的时候出现了网络波动(当然还有很多种情况，比如 Broker 意外重启等等)，这条回应没有发送成功。那么，消息队列没收到积分系统的回应会不会尝试重发这个消息？问题就来了，我再发这个消息，万一它又给 FrancisQ 的账户加上 500 积分怎么办呢？所以我们需要给我们的消费者实现 幂等 ，也就是对同一个消息的处理结果，执行多少次都不变。那么如何给业务实现幂等呢？这个还是需要结合具体的业务的。你可以使用 写入 Redis 来保证，因为 Redis 的 key 和 value 就是天然支持幂等的。当然还有使用 数据库插入法 ，基于数据库的唯一键来保证重复数据不会被插入多条。不过最主要的还是需要 根据特定场景使用特定的解决方案 ，你要知道你的消息消费是否是完全不可重复消费还是可以忍受重复消费的，然后再选择强校验和弱校验的方式。毕竟在 CS 领域还是很少有技术银弹的说法。而在整个互联网领域，幂等不仅仅适用于消息队列的重复消费问题，这些实现幂等的方法，也同样适用于，在其他场景中来解决重复请求或者重复调用的问题 。比如将 HTTP 服务设计成幂等的，解决前端或者 APP 重复提交表单数据的问题 ，也可以将一个微服务设计成幂等的，解决 RPC 框架自动重试导致的 重复调用问题 。RocketMQ 如何实现分布式事务？如何解释分布式事务呢？事务大家都知道吧？要么都执行要么都不执行 。在同一个系统中我们可以轻松地实现事务，但是在分布式架构中，我们有很多服务是部署在不同系统之间的，而不同服务之间又需要进行调用。比如此时我下订单然后增加积分，如果保证不了分布式事务的话，就会出现 A 系统下了订单，但是 B 系统增加积分失败或者 A 系统没有下订单，B 系统却增加了积分。前者对用户不友好，后者对运营商不利，这是我们都不愿意见到的。那么，如何去解决这个问题呢？如今比较常见的分布式事务实现有 2PC、TCC 和事务消息(half 半消息机制)。每一种实现都有其特定的使用场景，但是也有各自的问题，都不是完美的解决方案。在 RocketMQ 中使用的是 事务消息加上事务反查机制 来解决分布式事务问题的。我画了张图，大家可以对照着图进行理解。在第一步发送的 half 消息 ，它的意思是 在事务提交之前，对于消费者来说，这个消息是不可见的 。那么，如何做到写入消息但是对用户不可见呢？RocketMQ 事务消息的做法是：如果消息是 half 消息，将备份原消息的主题与消息消费队列，然后 改变主题 为 RMQ_SYS_TRANS_HALF_TOPIC。由于消费组未订阅该主题，故消费端无法消费 half 类型的消息，然后 RocketMQ 会开启一个定时任务，从 Topic 为 RMQ_SYS_TRANS_HALF_TOPIC 中拉取消息进行消费，根据生产者组获取一个服务提供者发送回查事务状态请求，根据事务状态来决定是提交或回滚消息。你可以试想一下，如果没有从第 5 步开始的 事务反查机制 ，如果出现网路波动第 4 步没有发送成功，这样就会产生 MQ 不知道是不是需要给消费者消费的问题，他就像一个无头苍蝇一样。在 RocketMQ 中就是使用的上述的事务反查来解决的，而在 Kafka 中通常是直接抛出一个异常让用户来自行解决。你还需要注意的是，在 MQ Server 指向系统 B 的操作已经和系统 A 不相关了，也就是说在消息队列中的分布式事务是——本地事务和存储消息到消息队列才是同一个事务。这样也就产生了事务的最终一致性，因为整个过程是异步的，每个系统只要保证它自己那一部分的事务就行了。实践中会遇到的问题：事务消息需要一个事务监听器来监听本地事务是否成功，并且事务监听器接口只允许被实现一次。那就意味着需要把各种事务消息的本地事务都写在一个接口方法里面，必将会产生大量的耦合和类型判断。采用函数 Function 接口来包装整个业务过程，作为一个参数传递到监听器的接口方法中。再调用 Function 的 apply() 方法来执行业务，事务也会在 apply() 方法中执行。让监听器与业务之间实现解耦，使之具备了真实生产环境中的可行性。1.模拟一个添加用户浏览记录的需求@PostMapping("/add") @ApiOperation("添加用户浏览记录") public Result<TransactionSendResult> add(Long userId, Long forecastLogId) { // 函数式编程:浏览记录入库 Function<String, Boolean> function = transactionId -> viewHistoryHandler.addViewHistory(transactionId, userId, forecastLogId); Map<String, Long> hashMap = new HashMap<>(); hashMap.put("userId", userId); hashMap.put("forecastLogId", forecastLogId); String jsonString = JSON.toJSONString(hashMap); // 发送事务消息;将本地的事务操作,用函数Function接口接收,作为一个参数传入到方法中 TransactionSendResult transactionSendResult = mqProducerService.sendTransactionMessage(jsonString, MQDestination.TAG_ADD_VIEW_HISTORY, function); return Result.success(transactionSendResult); }2.发送事务消息的方法/** * 发送事务消息 * * @param msgBody * @param tag * @param function * @return */ public TransactionSendResult sendTransactionMessage(String msgBody, String tag, Function<String, Boolean> function) { // 构建消息体 Message<String> message = buildMessage(msgBody); // 构建消息投递信息 String destination = buildDestination(tag); TransactionSendResult result = rocketMQTemplate.sendMessageInTransaction(destination, message, function); return result; }3.生产者消息监听器,只允许一个类去实现该监听器@Slf4j @RocketMQTransactionListener public class TransactionMsgListener implements RocketMQLocalTransactionListener { @Autowired private RedisService redisService; /** * 执行本地事务（在发送消息成功时执行） * * @param message * @param o * @return commit or rollback or unknown */ @Override public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) { // 1、获取事务ID String transactionId = null; try { transactionId = message.getHeaders().get("rocketmq_TRANSACTION_ID").toString(); // 2、判断传入函数对象是否为空，如果为空代表没有要执行的业务直接抛弃消息 if (o == null) { //返回ROLLBACK状态的消息会被丢弃 log.info("事务消息回滚，没有需要处理的业务 transactionId={}", transactionId); return RocketMQLocalTransactionState.ROLLBACK; } // 将Object o转换成Function对象 Function<String, Boolean> function = (Function<String, Boolean>) o; // 执行业务 事务也会在function.apply中执行 Boolean apply = function.apply(transactionId); if (apply) { log.info("事务提交，消息正常处理 transactionId={}", transactionId); //返回COMMIT状态的消息会立即被消费者消费到 return RocketMQLocalTransactionState.COMMIT; } } catch (Exception e) { log.info("出现异常 返回ROLLBACK transactionId={}", transactionId); return RocketMQLocalTransactionState.ROLLBACK; } return RocketMQLocalTransactionState.ROLLBACK; } /** * 事务回查机制，检查本地事务的状态 * * @param message * @return */ @Override public RocketMQLocalTransactionState checkLocalTransaction(Message message) { String transactionId = message.getHeaders().get("rocketmq_TRANSACTION_ID").toString(); // 查redis MqTransaction mqTransaction = redisService.getCacheObject("mqTransaction:" + transactionId); if (Objects.isNull(mqTransaction)) { return RocketMQLocalTransactionState.ROLLBACK; } return RocketMQLocalTransactionState.COMMIT; } }4.模拟的业务场景,这里的方法必须提取出来,放在别的类里面.如果调用方与被调用方在同一个类中,会发生事务失效的问题.@Component public class ViewHistoryHandler { @Autowired private IViewHistoryService viewHistoryService; @Autowired private IMqTransactionService mqTransactionService; @Autowired private RedisService redisService; /** * 浏览记录入库 * * @param transactionId * @param userId * @param forecastLogId * @return */ @Transactional public Boolean addViewHistory(String transactionId, Long userId, Long forecastLogId) { // 构建浏览记录 ViewHistory viewHistory = new ViewHistory(); viewHistory.setUserId(userId); viewHistory.setForecastLogId(forecastLogId); viewHistory.setCreateTime(LocalDateTime.now()); boolean save = viewHistoryService.save(viewHistory); // 本地事务信息 MqTransaction mqTransaction = new MqTransaction(); mqTransaction.setTransactionId(transactionId); mqTransaction.setCreateTime(new Date()); mqTransaction.setStatus(MqTransaction.StatusEnum.VALID.getStatus()); // 1.可以把事务信息存数据库 mqTransactionService.save(mqTransaction); // 2.也可以选择存redis,4个小时有效期,'4个小时'是RocketMQ内置的最大回查超时时长,过期未确认将强制回滚 redisService.setCacheObject("mqTransaction:" + transactionId, mqTransaction, 4L, TimeUnit.HOURS); // 放开注释,模拟异常,事务回滚 // int i = 10 / 0; return save; } }5.消费消息,以及幂等处理@Service @RocketMQMessageListener(topic = MQDestination.TOPIC, selectorExpression = MQDestination.TAG_ADD_VIEW_HISTORY, consumerGroup = MQDestination.TAG_ADD_VIEW_HISTORY) public class ConsumerAddViewHistory implements RocketMQListener<Message> { // 监听到消息就会执行此方法 @Override public void onMessage(Message message) { // 幂等校验 String transactionId = message.getTransactionId(); // 查redis MqTransaction mqTransaction = redisService.getCacheObject("mqTransaction:" + transactionId); // 不存在事务记录 if (Objects.isNull(mqTransaction)) { return; } // 已消费 if (Objects.equals(mqTransaction.getStatus(), MqTransaction.StatusEnum.CONSUMED.getStatus())) { return; } String msg = new String(message.getBody()); Map<String, Long> map = JSON.parseObject(msg, new TypeReference<HashMap<String, Long>>() { }); Long userId = map.get("userId"); Long forecastLogId = map.get("forecastLogId"); // 下游的业务处理 // TODO 记录用户喜好,更新用户画像 // TODO 更新'证券预测文章'的浏览量,重新计算文章的曝光排序 // 更新状态为已消费 mqTransaction.setUpdateTime(new Date()); mqTransaction.setStatus(MqTransaction.StatusEnum.CONSUMED.getStatus()); redisService.setCacheObject("mqTransaction:" + transactionId, mqTransaction, 4L, TimeUnit.HOURS); log.info("监听到消息：msg={}", JSON.toJSONString(map)); } }如何解决消息堆积问题？在上面我们提到了消息队列一个很重要的功能——削峰 。那么如果这个峰值太大了导致消息堆积在队列中怎么办呢？其实这个问题可以将它广义化，因为产生消息堆积的根源其实就只有两个——生产者生产太快或者消费者消费太慢。我们可以从多个角度去思考解决这个问题，当流量到峰值的时候是因为生产者生产太快，我们可以使用一些 限流降级 的方法，当然你也可以增加多个消费者实例去水平扩展增加消费能力来匹配生产的激增。如果消费者消费过慢的话，我们可以先检查 是否是消费者出现了大量的消费错误 ，或者打印一下日志查看是否是哪一个线程卡死，出现了锁资源不释放等等的问题。当然，最快速解决消息堆积问题的方法还是增加消费者实例，不过 同时你还需要增加每个主题的队列数量 。别忘了在 RocketMQ 中，一个队列只会被一个消费者消费 ，如果你仅仅是增加消费者实例就会出现我一开始给你画架构图的那种情况。什么是回溯消费？回溯消费是指 Consumer 已经消费成功的消息，由于业务上需求需要重新消费，在RocketMQ 中， Broker 在向Consumer 投递成功消息后，消息仍然需要保留 。并且重新消费一般是按照时间维度，例如由于 Consumer 系统故障，恢复后需要重新消费 1 小时前的数据，那么 Broker 要提供一种机制，可以按照时间维度来回退消费进度。RocketMQ 支持按照时间回溯消费，时间维度精确到毫秒。这是官方文档的解释，我直接照搬过来就当科普了 😁😁😁。RocketMQ 如何保证高性能读写传统 IO 方式传统的 IO 读写其实就是 read + write 的操作，整个过程会分为如下几步用户调用 read()方法，开始读取数据，此时发生一次上下文从用户态到内核态的切换，也就是图示的切换 1将磁盘数据通过 DMA 拷贝到内核缓存区将内核缓存区的数据拷贝到用户缓冲区，这样用户，也就是我们写的代码就能拿到文件的数据read()方法返回，此时就会从内核态切换到用户态，也就是图示的切换 2当我们拿到数据之后，就可以调用 write()方法，此时上下文会从用户态切换到内核态，即图示切换 3CPU 将用户缓冲区的数据拷贝到 Socket 缓冲区将 Socket 缓冲区数据拷贝至网卡write()方法返回，上下文重新从内核态切换到用户态，即图示切换 4整个过程发生了 4 次上下文切换和 4 次数据的拷贝，这在高并发场景下肯定会严重影响读写性能故引入了零拷贝技术零拷贝技术mmapmmap（memory map）是一种内存映射文件的方法，即将一个文件或者其它对象映射到进程的地址空间，实现文件磁盘地址和进程虚拟地址空间中一段虚拟地址的一一对映关系。简单地说就是内核缓冲区和应用缓冲区共享，从而减少了从读缓冲区到用户缓冲区的一次 CPU 拷贝。基于此上述架构图可变为：基于 mmap IO 读写其实就变成 mmap + write 的操作，也就是用 mmap 替代传统 IO 中的 read 操作。当用户发起 mmap 调用的时候会发生上下文切换 1，进行内存映射，然后数据被拷贝到内核缓冲区，mmap 返回，发生上下文切换 2；随后用户调用 write，发生上下文切换 3，将内核缓冲区的数据拷贝到 Socket 缓冲区，write 返回，发生上下文切换 4。发生 4 次上下文切换和 3 次 IO 拷贝操作，在 Java 中的实现：FileChannel fileChannel = new RandomAccessFile("test.txt", "rw").getChannel(); MappedByteBuffer mappedByteBuffer = fileChannel.map(FileChannel.MapMode.READ_WRITE, 0, fileChannel.size());sendfilesendfile()跟 mmap()一样，也会减少一次 CPU 拷贝，但是它同时也会减少两次上下文切换。如图，用户在发起 sendfile()调用时会发生切换 1，之后数据通过 DMA 拷贝到内核缓冲区，之后再将内核缓冲区的数据 CPU 拷贝到 Socket 缓冲区，最后拷贝到网卡，sendfile()返回，发生切换 2。发生了 3 次拷贝和两次切换。Java 也提供了相应 api：FileChannel channel = FileChannel.open(Paths.get("./test.txt"), StandardOpenOption.WRITE, StandardOpenOption.CREATE); //调用transferTo方法向目标数据传输 channel.transferTo(position, len, target);在如上代码中，并没有文件的读写操作，而是直接将文件的数据传输到 target 目标缓冲区，也就是说，sendfile 是无法知道文件的具体的数据的；但是 mmap 不一样，他是可以修改内核缓冲区的数据的。假设如果需要对文件的内容进行修改之后再传输，只有 mmap 可以满足。通过上面的一些介绍，结论是基于零拷贝技术，可以减少 CPU 的拷贝次数和上下文切换次数，从而可以实现文件高效的读写操作。RocketMQ 内部主要是使用基于 mmap 实现的零拷贝(其实就是调用上述提到的 api)，用来读写文件，这也是 RocketMQ 为什么快的一个很重要原因。RocketMQ 的刷盘机制上面我讲了那么多的 RocketMQ 的架构和设计原理，你有没有好奇在 Topic 中的 队列是以什么样的形式存在的？队列中的消息又是如何进行存储持久化的呢？我在上文中提到的 同步刷盘 和 异步刷盘 又是什么呢？它们会给持久化带来什么样的影响呢？下面我将给你们一一解释。同步刷盘和异步刷盘如上图所示，在同步刷盘中需要等待一个刷盘成功的 ACK ，同步刷盘对 MQ 消息可靠性来说是一种不错的保障，但是 性能上会有较大影响 ，一般地适用于金融等特定业务场景。而异步刷盘往往是开启一个线程去异步地执行刷盘操作。消息刷盘采用后台异步线程提交的方式进行， 降低了读写延迟 ，提高了 MQ 的性能和吞吐量，一般适用于如发验证码等对于消息保证要求不太高的业务场景。一般地，异步刷盘只有在 Broker 意外宕机的时候会丢失部分数据，你可以设置 Broker 的参数 FlushDiskType 来调整你的刷盘策略(ASYNC_FLUSH 或者 SYNC_FLUSH)。同步复制和异步复制上面的同步刷盘和异步刷盘是在单个结点层面的，而同步复制和异步复制主要是指的 Borker 主从模式下，主节点返回消息给客户端的时候是否需要同步从节点。同步复制：也叫 “同步双写”，也就是说，只有消息同步双写到主从节点上时才返回写入成功 。异步复制：消息写入主节点之后就直接返回写入成功 。然而，很多事情是没有完美的方案的，就比如我们进行消息写入的节点越多就更能保证消息的可靠性，但是随之的性能也会下降，所以需要程序员根据特定业务场景去选择适应的主从复制方案。那么，异步复制会不会也像异步刷盘那样影响消息的可靠性呢？答案是不会的，因为两者就是不同的概念，对于消息可靠性是通过不同的刷盘策略保证的，而像异步同步复制策略仅仅是影响到了 可用性 。为什么呢？其主要原因是 RocketMQ 是不支持自动主从切换的，当主节点挂掉之后，生产者就不能再给这个主节点生产消息了。比如这个时候采用异步复制的方式，在主节点还未发送完需要同步的消息的时候主节点挂掉了，这个时候从节点就少了一部分消息。但是此时生产者无法再给主节点生产消息了，消费者可以自动切换到从节点进行消费(仅仅是消费)，所以在主节点挂掉的时间只会产生主从结点短暂的消息不一致的情况，降低了可用性，而当主节点重启之后，从节点那部分未来得及复制的消息还会继续复制。在单主从架构中，如果一个主节点挂掉了，那么也就意味着整个系统不能再生产了。那么这个可用性的问题能否解决呢？一个主从不行那就多个主从的呗，别忘了在我们最初的架构图中，每个 Topic 是分布在不同 Broker 中的。但是这种复制方式同样也会带来一个问题，那就是无法保证 严格顺序 。在上文中我们提到了如何保证的消息顺序性是通过将一个语义的消息发送在同一个队列中，使用 Topic 下的队列来保证顺序性的。如果此时我们主节点 A 负责的是订单 A 的一系列语义消息，然后它挂了，这样其他节点是无法代替主节点 A 的，如果我们任意节点都可以存入任何消息，那就没有顺序性可言了。而在 RocketMQ 中采用了 Dledger 解决这个问题。他要求在写入消息的时候，要求至少消息复制到半数以上的节点之后，才给客⼾端返回写⼊成功，并且它是⽀持通过选举来动态切换主节点的。这里我就不展开说明了，读者可以自己去了解。也不是说 Dledger 是个完美的方案，至少在 Dledger 选举过程中是无法提供服务的，而且他必须要使用三个节点或以上，如果多数节点同时挂掉他也是无法保证可用性的，而且要求消息复制半数以上节点的效率和直接异步复制还是有一定的差距的。存储机制还记得上面我们一开始的三个问题吗？到这里第三个问题已经解决了。但是，在 Topic 中的 队列是以什么样的形式存在的？队列中的消息又是如何进行存储持久化的呢？ 还未解决，其实这里涉及到了 RocketMQ 是如何设计它的存储结构了。我首先想大家介绍 RocketMQ 消息存储架构中的三大角色——CommitLog、ConsumeQueue 和 IndexFile 。CommitLog：消息主体以及元数据的存储主体，存储 Producer 端写入的消息主体内容,消息内容不是定长的。单个文件大小默认 1G ，文件名长度为 20 位，左边补零，剩余为起始偏移量，比如 00000000000000000000 代表了第一个文件，起始偏移量为 0，文件大小为 1G=1073741824；当第一个文件写满了，第二个文件为 00000000001073741824，起始偏移量为 1073741824，以此类推。消息主要是顺序写入日志文件，当文件满了，写入下一个文件。ConsumeQueue：消息消费队列，引入的目的主要是提高消息消费的性能(我们再前面也讲了)，由于RocketMQ 是基于主题 Topic 的订阅模式，消息消费是针对主题进行的，如果要遍历 commitlog 文件中根据 Topic 检索消息是非常低效的。Consumer 即可根据 ConsumeQueue 来查找待消费的消息。其中，ConsumeQueue（逻辑消费队列）作为消费消息的索引，保存了指定 Topic 下的队列消息在 CommitLog 中的起始物理偏移量 offset ，消息大小 size 和消息 Tag 的 HashCode 值。consumequeue 文件可以看成是基于 topic 的 commitlog 索引文件，故 consumequeue 文件夹的组织方式如下：topic/queue/file 三层组织结构，具体存储路径为：$HOME/store/consumequeue/{topic}/{queueId}/{fileName}。同样 consumequeue 文件采取定长设计，每一个条目共 20 个字节，分别为 8 字节的 commitlog 物理偏移量、4 字节的消息长度、8 字节 tag hashcode，单个文件由 30W 个条目组成，可以像数组一样随机访问每一个条目，每个 ConsumeQueue文件大小约 5.72M；IndexFile：IndexFile（索引文件）提供了一种可以通过 key 或时间区间来查询消息的方法。这里只做科普不做详细介绍。总结来说，整个消息存储的结构，最主要的就是 CommitLoq 和 ConsumeQueue 。而 ConsumeQueue 你可以大概理解为 Topic 中的队列。RocketMQ 采用的是 混合型的存储结构 ，即为 Broker 单个实例下所有的队列共用一个日志数据文件来存储消息。有意思的是在同样高并发的 Kafka 中会为每个 Topic 分配一个存储文件。这就有点类似于我们有一大堆书需要装上书架，RockeMQ 是不分书的种类直接成批的塞上去的，而 Kafka 是将书本放入指定的分类区域的。而 RocketMQ 为什么要这么做呢？原因是 提高数据的写入效率 ，不分 Topic 意味着我们有更大的几率获取 成批 的消息进行数据写入，但也会带来一个麻烦就是读取消息的时候需要遍历整个大文件，这是非常耗时的。所以，在 RocketMQ 中又使用了 ConsumeQueue 作为每个队列的索引文件来 提升读取消息的效率。我们可以直接根据队列的消息序号，计算出索引的全局位置（索引序号*索引固定⻓度 20），然后直接读取这条索引，再根据索引中记录的消息的全局位置，找到消息。讲到这里，你可能对 RockeMQ 的存储架构还有些模糊，没事，我们结合着图来理解一下。emmm，是不是有一点复杂 🤣，看英文图片和英文文档的时候就不要怂，硬着头皮往下看就行。如果上面没看懂的读者一定要认真看下面的流程分析！首先，在最上面的那一块就是我刚刚讲的你现在可以直接 把 ConsumerQueue 理解为 Queue。在图中最左边说明了红色方块代表被写入的消息，虚线方块代表等待被写入的。左边的生产者发送消息会指定 Topic、QueueId 和具体消息内容，而在 Broker 中管你是哪门子消息，他直接 全部顺序存储到了 CommitLog。而根据生产者指定的 Topic 和 QueueId 将这条消息本身在 CommitLog 的偏移(offset)，消息本身大小，和 tag 的 hash 值存入对应的 ConsumeQueue 索引文件中。而在每个队列中都保存了 ConsumeOffset 即每个消费者组的消费位置(我在架构那里提到了，忘了的同学可以回去看一下)，而消费者拉取消息进行消费的时候只需要根据 ConsumeOffset 获取下一个未被消费的消息就行了。上述就是我对于整个消息存储架构的大概理解(这里不涉及到一些细节讨论，比如稀疏索引等等问题)，希望对你有帮助。因为有一个知识点因为写嗨了忘讲了，想想在哪里加也不好，所以我留给大家去思考 🤔🤔 一下吧。为什么 CommitLog 文件要设计成固定大小的长度呢？提醒：内存映射机制。总结总算把这篇博客写完了。我讲的你们还记得吗 😅？这篇文章中我主要想大家介绍了消息队列出现的原因消息队列的作用(异步，解耦，削峰)消息队列带来的一系列问题(消息堆积、重复消费、顺序消费、分布式事务等等)消息队列的两种消息模型——队列和主题模式分析了 RocketMQ 的技术架构(NameServer、Broker、Producer、Consumer)结合 RocketMQ 回答了消息队列副作用的解决方案介绍了 RocketMQ 的存储机制和刷盘策略。
AI 摘要 文章系统梳理了 Redis 九大数据类型的底层结构、常用命令与典型场景：String 做缓存/计数/锁/Session，List 做队列，Hash 存对象/购物车，Set 去重与交集运算，Zset 排行榜，BitMap 二值统计，HyperLogLog 海量基数，GEO 附近位置，Stream 专业队列；并给出选型与内存优化建议。 此摘要根据文章内容自动生成，仅用于文章内容的解释与总结Redis 常见数据类型和应用场景大家好，我是樱舞。我们都知道 Redis 提供了丰富的数据类型，常见的有五种：String（字符串），Hash（哈希），List（列表），Set（集合）、Zset（有序集合）。随着 Redis 版本的更新，后面又支持了四种数据类型： BitMap（2.2 版新增）、HyperLogLog（2.8 版新增）、GEO（3.2 版新增）、Stream（5.0 版新增）。每种数据对象都各自的应用场景，你能说出它们各自的应用场景吗？面试过程中，这个问题也很常被问到，又比如会举例一个应用场景来问你，让你说使用哪种 Redis 数据类型来实现。所以，这次我们就来学习 Redis 数据类型的使用以及应用场景。PS：你可以自己本机安装 Redis 或者通过 Redis 官网提供的在线 Redis 环境 来敲命令。String介绍String 是最基本的 key-value 结构，key 是唯一标识，value 是具体的值，value 其实不仅是字符串，也可以是数字（整数或浮点数），value 最多可以容纳的数据长度是 512M。内部实现String 类型的底层的数据结构实现主要是 int 和 SDS（简单动态字符串）。SDS 和我们认识的 C 字符串不太一样，之所以没有使用 C 语言的字符串表示，因为 SDS 相比于 C 的原生字符串：SDS 不仅可以保存文本数据，还可以保存二进制数据。因为 SDS 使用 len 属性的值而不是空字符来判断字符串是否结束，并且 SDS 的所有 API 都会以处理二进制的方式来处理 SDS 存放在 buf[] 数组里的数据。所以 SDS 不光能存放文本数据，而且能保存图片、音频、视频、压缩文件这样的二进制数据。SDS 获取字符串长度的时间复杂度是 O(1)。因为 C 语言的字符串并不记录自身长度，所以获取长度的复杂度为 O(n)；而 SDS 结构里用 len 属性记录了字符串长度，所以复杂度为 O(1)。Redis 的 SDS API 是安全的，拼接字符串不会造成缓冲区溢出。因为 SDS 在拼接字符串之前会检查 SDS 空间是否满足要求，如果空间不够会自动扩容，所以不会导致缓冲区溢出的问题。字符串对象的内部编码（encoding）有 3 种：int、raw 和 embstr。如果一个字符串对象保存的是整数值，并且这个整数值可以用long类型来表示，那么字符串对象会将整数值保存在字符串对象结构的ptr属性里面（将void*转换成 long），并将字符串对象的编码设置为int。如果字符串对象保存的是一个字符串，并且这个字符串的长度小于等于 32 字节（redis 2.+版本），那么字符串对象将使用一个简单动态字符串（SDS）来保存这个字符串，并将对象的编码设置为embstr， embstr编码是专门用于保存短字符串的一种优化编码方式：如果字符串对象保存的是一个字符串，并且这个字符串的长度大于 32 字节（redis 2.+版本），那么字符串对象将使用一个简单动态字符串（SDS）来保存这个字符串，并将对象的编码设置为raw：注意，embstr 编码和 raw 编码的边界在 redis 不同版本中是不一样的：redis 2.+ 是 32 字节redis 3.0-4.0 是 39 字节redis 5.0 是 44 字节可以看到embstr和raw编码都会使用SDS来保存值，但不同之处在于embstr会通过一次内存分配函数来分配一块连续的内存空间来保存redisObject和SDS，而raw编码会通过调用两次内存分配函数来分别分配两块空间来保存redisObject和SDS。Redis 这样做会有很多好处：embstr编码将创建字符串对象所需的内存分配次数从 raw 编码的两次降低为一次；释放 embstr编码的字符串对象同样只需要调用一次内存释放函数；因为embstr编码的字符串对象的所有数据都保存在一块连续的内存里面可以更好的利用 CPU 缓存提升性能。但是 embstr 也有缺点的：如果字符串的长度增加需要重新分配内存时，整个 redisObject 和 sds 都需要重新分配空间，所以embstr 编码的字符串对象实际上是只读的，redis 没有为 embstr 编码的字符串对象编写任何相应的修改程序。当我们对 embstr 编码的字符串对象执行任何修改命令（例如 append）时，程序会先将对象的编码从 embstr 转换成 raw，然后再执行修改命令。常用指令普通字符串的基本操作：# 设置 key-value 类型的值 > SET name lin OK # 根据 key 获得对应的 value > GET name "lin" # 判断某个 key 是否存在 > EXISTS name (integer) 1 # 返回 key 所储存的字符串值的长度 > STRLEN name (integer) 3 # 删除某个 key 对应的值 > DEL name (integer) 1批量设置 :# 批量设置 key-value 类型的值 > MSET key1 value1 key2 value2 OK # 批量获取多个 key 对应的 value > MGET key1 key2 1) "value1" 2) "value2"计数器（字符串的内容为整数的时候可以使用）：# 设置 key-value 类型的值 > SET number 0 OK # 将 key 中储存的数字值增一 > INCR number (integer) 1 # 将key中存储的数字值加 10 > INCRBY number 10 (integer) 11 # 将 key 中储存的数字值减一 > DECR number (integer) 10 # 将key中存储的数字值键 10 > DECRBY number 10 (integer) 0过期（默认为永不过期）：# 设置 key 在 60 秒后过期（该方法是针对已经存在的key设置过期时间） > EXPIRE name 60 (integer) 1 # 查看数据还有多久过期 > TTL name (integer) 51 #设置 key-value 类型的值，并设置该key的过期时间为 60 秒 > SET key value EX 60 OK > SETEX key 60 value OK不存在就插入：# 不存在就插入（not exists） >SETNX key value (integer) 1应用场景缓存对象使用 String 来缓存对象有两种方式：直接缓存整个对象的 JSON，命令例子： SET user:1 '{"name":"xiaolin", "age":18}'。采用将 key 进行分离为 user:ID:属性，采用 MSET 存储，用 MGET 获取各属性值，命令例子： MSET user:1:name xiaolin user:1:age 18 user:2:name xiaomei user:2:age 20。常规计数因为 Redis 处理命令是单线程，所以执行命令的过程是原子的。因此 String 数据类型适合计数场景，比如计算访问次数、点赞、转发、库存数量等等。比如计算文章的阅读量：# 初始化文章的阅读量 > SET aritcle:readcount:1001 0 OK #阅读量+1 > INCR aritcle:readcount:1001 (integer) 1 #阅读量+1 > INCR aritcle:readcount:1001 (integer) 2 #阅读量+1 > INCR aritcle:readcount:1001 (integer) 3 # 获取对应文章的阅读量 > GET aritcle:readcount:1001 "3"分布式锁SET 命令有个 NX 参数可以实现「key 不存在才插入」，可以用它来实现分布式锁：如果 key 不存在，则显示插入成功，可以用来表示加锁成功；如果 key 存在，则会显示插入失败，可以用来表示加锁失败。一般而言，还会对分布式锁加上过期时间，分布式锁的命令如下：SET lock_key unique_value NX PX 10000lock_key 就是 key 键；unique_value 是客户端生成的唯一的标识；NX 代表只在 lock_key 不存在时，才对 lock_key 进行设置操作；PX 10000 表示设置 lock_key 的过期时间为 10s，这是为了避免客户端发生异常而无法释放锁。而解锁的过程就是将 lock_key 键删除，但不能乱删，要保证执行操作的客户端就是加锁的客户端。所以，解锁的时候，我们要先判断锁的 unique_value 是否为加锁客户端，是的话，才将 lock_key 键删除。可以看到，解锁是有两个操作，这时就需要 Lua 脚本来保证解锁的原子性，因为 Redis 在执行 Lua 脚本时，可以以原子性的方式执行，保证了锁释放操作的原子性。// 释放锁时，先比较 unique_value 是否相等，避免锁的误释放 if redis.call("get",KEYS[1]) == ARGV[1] then return redis.call("del",KEYS[1]) else return 0 end这样一来，就通过使用 SET 命令和 Lua 脚本在 Redis 单节点上完成了分布式锁的加锁和解锁。共享 Session 信息通常我们在开发后台管理系统时，会使用 Session 来保存用户的会话 (登录) 状态，这些 Session 信息会被保存在服务器端，但这只适用于单系统应用，如果是分布式系统此模式将不再适用。例如用户一的 Session 信息被存储在服务器一，但第二次访问时用户一被分配到服务器二，这个时候服务器并没有用户一的 Session 信息，就会出现需要重复登录的问题，问题在于分布式系统每次会把请求随机分配到不同的服务器。分布式系统单独存储 Session 流程图：因此，我们需要借助 Redis 对这些 Session 信息进行统一的存储和管理，这样无论请求发送到那台服务器，服务器都会去同一个 Redis 获取相关的 Session 信息，这样就解决了分布式系统下 Session 存储的问题。分布式系统使用同一个 Redis 存储 Session 流程图：List介绍List 列表是简单的字符串列表，按照插入顺序排序，可以从头部或尾部向 List 列表添加元素。列表的最大长度为 2^32 - 1，也即每个列表支持超过 40 亿个元素。内部实现List 类型的底层数据结构是由双向链表或压缩列表实现的：如果列表的元素个数小于 512 个（默认值，可由 list-max-ziplist-entries 配置），列表每个元素的值都小于 64 字节（默认值，可由 list-max-ziplist-value 配置），Redis 会使用压缩列表作为 List 类型的底层数据结构；如果列表的元素不满足上面的条件，Redis 会使用双向链表作为 List 类型的底层数据结构；但是在 Redis 3.2 版本之后，List 数据类型底层数据结构就只由 quicklist 实现了，替代了双向链表和压缩列表。常用命令# 将一个或多个值value插入到key列表的表头(最左边)，最后的值在最前面 LPUSH key value [value ...] # 将一个或多个值value插入到key列表的表尾(最右边) RPUSH key value [value ...] # 移除并返回key列表的头元素 LPOP key # 移除并返回key列表的尾元素 RPOP key # 返回列表key中指定区间内的元素，区间以偏移量start和stop指定，从0开始 LRANGE key start stop # 从key列表表头弹出一个元素，没有就阻塞timeout秒，如果timeout=0则一直阻塞 BLPOP key [key ...] timeout # 从key列表表尾弹出一个元素，没有就阻塞timeout秒，如果timeout=0则一直阻塞 BRPOP key [key ...] timeout应用场景消息队列消息队列在存取消息时，必须要满足三个需求，分别是消息保序、处理重复的消息和保证消息可靠性。Redis 的 List 和 Stream 两种数据类型，就可以满足消息队列的这三个需求。我们先来了解下基于 List 的消息队列实现方法，后面在介绍 Stream 数据类型时候，在详细说说 Stream。1、如何满足消息保序需求？List 本身就是按先进先出的顺序对数据进行存取的，所以，如果使用 List 作为消息队列保存消息的话，就已经能满足消息保序的需求了。List 可以使用 LPUSH + RPOP（或者反过来，RPUSH+LPOP）命令实现消息队列。生产者使用 LPUSH key value[value...] 将消息插入到队列的头部，如果 key 不存在则会创建一个空的队列再插入消息。消费者使用 RPOP key 依次读取队列的消息，先进先出。不过，在消费者读取数据时，有一个潜在的性能风险点。在生产者往 List 中写入数据时，List 并不会主动地通知消费者有新消息写入，如果消费者想要及时处理消息，就需要在程序中不停地调用 RPOP 命令（比如使用一个 while(1) 循环）。如果有新消息写入，RPOP 命令就会返回结果，否则，RPOP 命令返回空值，再继续循环。所以，即使没有新消息写入 List，消费者也要不停地调用 RPOP 命令，这就会导致消费者程序的 CPU 一直消耗在执行 RPOP 命令上，带来不必要的性能损失。为了解决这个问题，Redis 提供了 BRPOP 命令。BRPOP 命令也称为阻塞式读取，客户端在没有读到队列数据时，自动阻塞，直到有新的数据写入队列，再开始读取新数据。和消费者程序自己不停地调用 RPOP 命令相比，这种方式能节省 CPU 开销。2、如何处理重复的消息？消费者要实现重复消息的判断，需要 2 个方面的要求：每个消息都有一个全局的 ID。消费者要记录已经处理过的消息的 ID。当收到一条消息后，消费者程序就可以对比收到的消息 ID 和记录的已处理过的消息 ID，来判断当前收到的消息有没有经过处理。如果已经处理过，那么，消费者程序就不再进行处理了。但是 List 并不会为每个消息生成 ID 号，所以我们需要自行为每个消息生成一个全局唯一 ID，生成之后，我们在用 LPUSH 命令把消息插入 List 时，需要在消息中包含这个全局唯一 ID。例如，我们执行以下命令，就把一条全局 ID 为 111000102、库存量为 99 的消息插入了消息队列：> LPUSH mq "111000102:stock:99" (integer) 13、如何保证消息可靠性？当消费者程序从 List 中读取一条消息后，List 就不会再留存这条消息了。所以，如果消费者程序在处理消息的过程出现了故障或宕机，就会导致消息没有处理完成，那么，消费者程序再次启动后，就没法再次从 List 中读取消息了。为了留存消息，List 类型提供了 BRPOPLPUSH 命令，这个命令的作用是让消费者程序从一个 List 中读取消息，同时，Redis 会把这个消息再插入到另一个 List（可以叫作备份 List）留存。这样一来，如果消费者程序读了消息但没能正常处理，等它重启后，就可以从备份 List 中重新读取消息并进行处理了。好了，到这里可以知道基于 List 类型的消息队列，满足消息队列的三大需求（消息保序、处理重复的消息和保证消息可靠性）。消息保序：使用 LPUSH + RPOP；阻塞读取：使用 BRPOP；重复消息处理：生产者自行实现全局唯一 ID；消息的可靠性：使用 BRPOPLPUSHList 作为消息队列有什么缺陷？List 不支持多个消费者消费同一条消息，因为一旦消费者拉取一条消息后，这条消息就从 List 中删除了，无法被其它消费者再次消费。要实现一条消息可以被多个消费者消费，那么就要将多个消费者组成一个消费组，使得多个消费者可以消费同一条消息，但是 List 类型并不支持消费组的实现。这就要说起 Redis 从 5.0 版本开始提供的 Stream 数据类型了，Stream 同样能够满足消息队列的三大需求，而且它还支持「消费组」形式的消息读取。Hash介绍Hash 是一个键值对（key - value）集合，其中 value 的形式如： value=[{field1，value1}，...{fieldN，valueN}]。Hash 特别适合用于存储对象。 Hash 与 String 对象的区别如下图所示：内部实现Hash 类型的底层数据结构是由压缩列表或哈希表实现的：如果哈希类型元素个数小于 512 个（默认值，可由 hash-max-ziplist-entries 配置），所有值小于 64 字节（默认值，可由 hash-max-ziplist-value 配置）的话，Redis 会使用压缩列表作为 Hash 类型的底层数据结构；如果哈希类型元素不满足上面条件，Redis 会使用哈希表作为 Hash 类型的 底层数据结构。在 Redis 7.0 中，压缩列表数据结构已经废弃了，交由 listpack 数据结构来实现了。常用命令# 存储一个哈希表key的键值 HSET key field value # 获取哈希表key对应的field键值 HGET key field # 在一个哈希表key中存储多个键值对 HMSET key field value [field value...] # 批量获取哈希表key中多个field键值 HMGET key field [field ...] # 删除哈希表key中的field键值 HDEL key field [field ...] # 返回哈希表key中field的数量 HLEN key # 返回哈希表key中所有的键值 HGETALL key # 为哈希表key中field键的值加上增量n HINCRBY key field n 应用场景缓存对象Hash 类型的（key，field，value）的结构与对象的（对象 id，属性，值）的结构相似，也可以用来存储对象。我们以用户信息为例，它在关系型数据库中的结构是这样的：我们可以使用如下命令，将用户对象的信息存储到 Hash 类型：# 存储一个哈希表uid:1的键值 > HMSET uid:1 name Tom age 15 2 # 存储一个哈希表uid:2的键值 > HMSET uid:2 name Jerry age 13 2 # 获取哈希表用户id为1中所有的键值 > HGETALL uid:1 1) "name" 2) "Tom" 3) "age" 4) "15"Redis Hash 存储其结构如下图：在介绍 String 类型的应用场景时有所介绍，String + Json 也是存储对象的一种方式，那么存储对象时，到底用 String + json 还是用 Hash 呢？一般对象用 String + Json 存储，对象中某些频繁变化的属性可以考虑抽出来用 Hash 类型存储。购物车以用户 id 为 key，商品 id 为 field，商品数量为 value，恰好构成了购物车的 3 个要素，如下图所示。涉及的命令如下：添加商品：HSET cart:{用户id} {商品id} 1添加数量：HINCRBY cart:{用户id} {商品id} 1商品总数：HLEN cart:{用户id}删除商品：HDEL cart:{用户id} {商品id}获取购物车所有商品：HGETALL cart:{用户id}当前仅仅是将商品 ID 存储到了 Redis 中，在回显商品具体信息的时候，还需要拿着商品 id 查询一次数据库，获取完整的商品的信息。Set介绍Set 类型是一个无序并唯一的键值集合，它的存储顺序不会按照插入的先后顺序进行存储。一个集合最多可以存储 2^32-1 个元素。概念和数学中个的集合基本类似，可以交集，并集，差集等等，所以 Set 类型除了支持集合内的增删改查，同时还支持多个集合取交集、并集、差集。Set 类型和 List 类型的区别如下：List 可以存储重复元素，Set 只能存储非重复元素；List 是按照元素的先后顺序存储元素的，而 Set 则是无序方式存储元素的。内部实现Set 类型的底层数据结构是由哈希表或整数集合实现的：如果集合中的元素都是整数且元素个数小于 512 （默认值，set-maxintset-entries配置）个，Redis 会使用整数集合作为 Set 类型的底层数据结构；如果集合中的元素不满足上面条件，则 Redis 使用哈希表作为 Set 类型的底层数据结构。常用命令Set 常用操作：# 往集合key中存入元素，元素存在则忽略，若key不存在则新建 SADD key member [member ...] # 从集合key中删除元素 SREM key member [member ...] # 获取集合key中所有元素 SMEMBERS key # 获取集合key中的元素个数 SCARD key # 判断member元素是否存在于集合key中 SISMEMBER key member # 从集合key中随机选出count个元素，元素不从key中删除 SRANDMEMBER key [count] # 从集合key中随机选出count个元素，元素从key中删除 SPOP key [count]Set 运算操作：# 交集运算 SINTER key [key ...] # 将交集结果存入新集合destination中 SINTERSTORE destination key [key ...] # 并集运算 SUNION key [key ...] # 将并集结果存入新集合destination中 SUNIONSTORE destination key [key ...] # 差集运算 SDIFF key [key ...] # 将差集结果存入新集合destination中 SDIFFSTORE destination key [key ...]应用场景集合的主要几个特性，无序、不可重复、支持并交差等操作。因此 Set 类型比较适合用来数据去重和保障数据的唯一性，还可以用来统计多个集合的交集、差集和并集等，当我们存储的数据是无序并且需要去重的情况下，比较适合使用集合类型进行存储。但是要提醒你一下，这里有一个潜在的风险。Set 的差集、并集和交集的计算复杂度较高，在数据量较大的情况下，如果直接执行这些计算，会导致 Redis 实例阻塞。在主从集群中，为了避免主库因为 Set 做聚合计算（交集、差集、并集）时导致主库被阻塞，我们可以选择一个从库完成聚合统计，或者把数据返回给客户端，由客户端来完成聚合统计。点赞Set 类型可以保证一个用户只能点一个赞，这里举例子一个场景，key 是文章 id，value 是用户 id。uid:1 、uid:2、uid:3 三个用户分别对 article:1 文章点赞了。# uid:1 用户对文章 article:1 点赞 > SADD article:1 uid:1 (integer) 1 # uid:2 用户对文章 article:1 点赞 > SADD article:1 uid:2 (integer) 1 # uid:3 用户对文章 article:1 点赞 > SADD article:1 uid:3 (integer) 1uid:1 取消了对 article:1 文章点赞。> SREM article:1 uid:1 (integer) 1获取 article:1 文章所有点赞用户 :> SMEMBERS article:1 1) "uid:3" 2) "uid:2"获取 article:1 文章的点赞用户数量：> SCARD article:1 (integer) 2判断用户 uid:1 是否对文章 article:1 点赞了：> SISMEMBER article:1 uid:1 (integer) 0 # 返回0说明没点赞，返回1则说明点赞了共同关注Set 类型支持交集运算，所以可以用来计算共同关注的好友、公众号等。key 可以是用户 id，value 则是已关注的公众号的 id。uid:1 用户关注公众号 id 为 5、6、7、8、9，uid:2 用户关注公众号 id 为 7、8、9、10、11。# uid:1 用户关注公众号 id 为 5、6、7、8、9 > SADD uid:1 5 6 7 8 9 (integer) 5 # uid:2 用户关注公众号 id 为 7、8、9、10、11 > SADD uid:2 7 8 9 10 11 (integer) 5uid:1 和 uid:2 共同关注的公众号：# 获取共同关注 > SINTER uid:1 uid:2 1) "7" 2) "8" 3) "9"给 uid:2 推荐 uid:1 关注的公众号：> SDIFF uid:1 uid:2 1) "5" 2) "6"验证某个公众号是否同时被 uid:1 或 uid:2 关注：> SISMEMBER uid:1 5 (integer) 1 # 返回1，说明关注了 > SISMEMBER uid:2 5 (integer) 0 # 返回0，说明没关注抽奖活动存储某活动中中奖的用户名，Set 类型因为有去重功能，可以保证同一个用户不会中奖两次。key 为抽奖活动名，value 为员工名称，把所有员工名称放入抽奖箱：>SADD lucky Tom Jerry John Sean Marry Lindy Sary Mark (integer) 5如果允许重复中奖，可以使用 SRANDMEMBER 命令。# 抽取 1 个一等奖： > SRANDMEMBER lucky 1 1) "Tom" # 抽取 2 个二等奖： > SRANDMEMBER lucky 2 1) "Mark" 2) "Jerry" # 抽取 3 个三等奖： > SRANDMEMBER lucky 3 1) "Sary" 2) "Tom" 3) "Jerry"如果不允许重复中奖，可以使用 SPOP 命令。# 抽取一等奖1个 > SPOP lucky 1 1) "Sary" # 抽取二等奖2个 > SPOP lucky 2 1) "Jerry" 2) "Mark" # 抽取三等奖3个 > SPOP lucky 3 1) "John" 2) "Sean" 3) "Lindy"Zset介绍Zset 类型（有序集合类型）相比于 Set 类型多了一个排序属性 score（分值），对于有序集合 ZSet 来说，每个存储元素相当于有两个值组成的，一个是有序结合的元素值，一个是排序值。有序集合保留了集合不能有重复成员的特性（分值可以重复），但不同的是，有序集合中的元素可以排序。内部实现Zset 类型的底层数据结构是由压缩列表或跳表实现的：如果有序集合的元素个数小于 128 个，并且每个元素的值小于 64 字节时，Redis 会使用压缩列表作为 Zset 类型的底层数据结构；如果有序集合的元素不满足上面的条件，Redis 会使用跳表作为 Zset 类型的底层数据结构；在 Redis 7.0 中，压缩列表数据结构已经废弃了，交由 listpack 数据结构来实现了。常用命令Zset 常用操作：# 往有序集合key中加入带分值元素 ZADD key score member [[score member]...] # 往有序集合key中删除元素 ZREM key member [member...] # 返回有序集合key中元素member的分值 ZSCORE key member # 返回有序集合key中元素个数 ZCARD key # 为有序集合key中元素member的分值加上increment ZINCRBY key increment member # 正序获取有序集合key从start下标到stop下标的元素 ZRANGE key start stop [WITHSCORES] # 倒序获取有序集合key从start下标到stop下标的元素 ZREVRANGE key start stop [WITHSCORES] # 返回有序集合中指定分数区间内的成员，分数由低到高排序。 ZRANGEBYSCORE key min max [WITHSCORES] [LIMIT offset count] # 返回指定成员区间内的成员，按字典正序排列, 分数必须相同。 ZRANGEBYLEX key min max [LIMIT offset count] # 返回指定成员区间内的成员，按字典倒序排列, 分数必须相同 ZREVRANGEBYLEX key max min [LIMIT offset count]Zset 运算操作（相比于 Set 类型，ZSet 类型没有支持差集运算）：# 并集计算(相同元素分值相加)，numberkeys一共多少个key，WEIGHTS每个key对应的分值乘积 ZUNIONSTORE destkey numberkeys key [key...] # 交集计算(相同元素分值相加)，numberkeys一共多少个key，WEIGHTS每个key对应的分值乘积 ZINTERSTORE destkey numberkeys key [key...]应用场景Zset 类型（Sorted Set，有序集合）可以根据元素的权重来排序，我们可以自己来决定每个元素的权重值。比如说，我们可以根据元素插入 Sorted Set 的时间确定权重值，先插入的元素权重小，后插入的元素权重大。在面对需要展示最新列表、排行榜等场景时，如果数据更新频繁或者需要分页显示，可以优先考虑使用 Sorted Set。排行榜有序集合比较典型的使用场景就是排行榜。例如学生成绩的排名榜、游戏积分排行榜、视频播放排名、电商系统中商品的销量排名等。我们以博文点赞排名为例，樱舞发表了五篇博文，分别获得赞为 200、40、100、50、150。# arcticle:1 文章获得了200个赞 > ZADD user:xiaolin:ranking 200 arcticle:1 (integer) 1 # arcticle:2 文章获得了40个赞 > ZADD user:xiaolin:ranking 40 arcticle:2 (integer) 1 # arcticle:3 文章获得了100个赞 > ZADD user:xiaolin:ranking 100 arcticle:3 (integer) 1 # arcticle:4 文章获得了50个赞 > ZADD user:xiaolin:ranking 50 arcticle:4 (integer) 1 # arcticle:5 文章获得了150个赞 > ZADD user:xiaolin:ranking 150 arcticle:5 (integer) 1文章 arcticle:4 新增一个赞，可以使用 ZINCRBY 命令（为有序集合 key 中元素 member 的分值加上 increment）：> ZINCRBY user:xiaolin:ranking 1 arcticle:4 "51"查看某篇文章的赞数，可以使用 ZSCORE 命令（返回有序集合 key 中元素个数）：> ZSCORE user:xiaolin:ranking arcticle:4 "50"获取樱舞文章赞数最多的 3 篇文章，可以使用 ZREVRANGE 命令（倒序获取有序集合 key 从 start 下标到 stop 下标的元素）：# WITHSCORES 表示把 score 也显示出来 > ZREVRANGE user:xiaolin:ranking 0 2 WITHSCORES 1) "arcticle:1" 2) "200" 3) "arcticle:5" 4) "150" 5) "arcticle:3" 6) "100"获取樱舞 100 赞到 200 赞的文章，可以使用 ZRANGEBYSCORE 命令（返回有序集合中指定分数区间内的成员，分数由低到高排序）：> ZRANGEBYSCORE user:xiaolin:ranking 100 200 WITHSCORES 1) "arcticle:3" 2) "100" 3) "arcticle:5" 4) "150" 5) "arcticle:1" 6) "200"电话、姓名排序使用有序集合的 ZRANGEBYLEX 或 ZREVRANGEBYLEX 可以帮助我们实现电话号码或姓名的排序，我们以 ZRANGEBYLEX （返回指定成员区间内的成员，按 key 正序排列，分数必须相同）为例。注意：不要在分数不一致的 SortSet 集合中去使用 ZRANGEBYLEX 和 ZREVRANGEBYLEX 指令，因为获取的结果会不准确。1、电话排序我们可以将电话号码存储到 SortSet 中，然后根据需要来获取号段：> ZADD phone 0 13100111100 0 13110114300 0 13132110901 (integer) 3 > ZADD phone 0 13200111100 0 13210414300 0 13252110901 (integer) 3 > ZADD phone 0 13300111100 0 13310414300 0 13352110901 (integer) 3获取所有号码：> ZRANGEBYLEX phone - + 1) "13100111100" 2) "13110114300" 3) "13132110901" 4) "13200111100" 5) "13210414300" 6) "13252110901" 7) "13300111100" 8) "13310414300" 9) "13352110901"获取 132 号段的号码：> ZRANGEBYLEX phone [132 (133 1) "13200111100" 2) "13210414300" 3) "13252110901"获取 132、133 号段的号码：> ZRANGEBYLEX phone [132 (134 1) "13200111100" 2) "13210414300" 3) "13252110901" 4) "13300111100" 5) "13310414300" 6) "13352110901"2、姓名排序> zadd names 0 Toumas 0 Jake 0 Bluetuo 0 Gaodeng 0 Aimini 0 Aidehua (integer) 6获取所有人的名字：> ZRANGEBYLEX names - + 1) "Aidehua" 2) "Aimini" 3) "Bluetuo" 4) "Gaodeng" 5) "Jake" 6) "Toumas"获取名字中大写字母 A 开头的所有人：> ZRANGEBYLEX names [A (B 1) "Aidehua" 2) "Aimini"获取名字中大写字母 C 到 Z 的所有人：> ZRANGEBYLEX names [C [Z 1) "Gaodeng" 2) "Jake" 3) "Toumas"BitMap介绍Bitmap，即位图，是一串连续的二进制数组（0 和 1），可以通过偏移量（offset）定位元素。BitMap 通过最小的单位 bit 来进行0|1的设置，表示某个元素的值或者状态，时间复杂度为 O(1)。由于 bit 是计算机中最小的单位，使用它进行储存将非常节省空间，特别适合一些数据量大且使用二值统计的场景。内部实现Bitmap 本身是用 String 类型作为底层数据结构实现的一种统计二值状态的数据类型。String 类型是会保存为二进制的字节数组，所以，Redis 就把字节数组的每个 bit 位利用起来，用来表示一个元素的二值状态，你可以把 Bitmap 看作是一个 bit 数组。常用命令bitmap 基本操作：# 设置值，其中value只能是 0 和 1 SETBIT key offset value # 获取值 GETBIT key offset # 获取指定范围内值为 1 的个数 # start 和 end 以字节为单位 BITCOUNT key start endbitmap 运算操作：# BitMap间的运算 # operations 位移操作符，枚举值 AND 与运算 & OR 或运算 | XOR 异或 ^ NOT 取反 ~ # result 计算的结果，会存储在该key中 # key1 … keyn 参与运算的key，可以有多个，空格分割，not运算只能一个key # 当 BITOP 处理不同长度的字符串时，较短的那个字符串所缺少的部分会被看作 0。返回值是保存到 destkey 的字符串的长度（以字节byte为单位），和输入 key 中最长的字符串长度相等。 BITOP [operations] [result] [key1] [keyn…] # 返回指定key中第一次出现指定value(0/1)的位置 BITPOS [key] [value]应用场景Bitmap 类型非常适合二值状态统计的场景，这里的二值状态就是指集合元素的取值就只有 0 和 1 两种，在记录海量数据时，Bitmap 能够有效地节省内存空间。签到统计在签到打卡的场景中，我们只用记录签到（1）或未签到（0），所以它就是非常典型的二值状态。签到统计时，每个用户一天的签到用 1 个 bit 位就能表示，一个月（假设是 31 天）的签到情况用 31 个 bit 位就可以，而一年的签到也只需要用 365 个 bit 位，根本不用太复杂的集合类型。假设我们要统计 ID 100 的用户在 2022 年 6 月份的签到情况，就可以按照下面的步骤进行操作。第一步，执行下面的命令，记录该用户 6 月 3 号已签到。SETBIT uid:sign:100:202206 2 1第二步，检查该用户 6 月 3 日是否签到。GETBIT uid:sign:100:202206 2 第三步，统计该用户在 6 月份的签到次数。BITCOUNT uid:sign:100:202206这样，我们就知道该用户在 6 月份的签到情况了。如何统计这个月首次打卡时间呢？Redis 提供了 BITPOS key bitValue [start] [end]指令，返回数据表示 Bitmap 中第一个值为 bitValue 的 offset 位置。在默认情况下，命令将检测整个位图，用户可以通过可选的 start 参数和 end 参数指定要检测的范围。所以我们可以通过执行这条命令来获取 userID = 100 在 2022 年 6 月份首次打卡日期：BITPOS uid:sign:100:202206 1需要注意的是，因为 offset 从 0 开始的，所以我们需要将返回的 value + 1。判断用户登陆态Bitmap 提供了 GETBIT、SETBIT 操作，通过一个偏移值 offset 对 bit 数组的 offset 位置的 bit 位进行读写操作，需要注意的是 offset 从 0 开始。只需要一个 key = login_status 表示存储用户登陆状态集合数据，将用户 ID 作为 offset，在线就设置为 1，下线设置 0。通过 GETBIT判断对应的用户是否在线。50000 万 用户只需要 6 MB 的空间。假如我们要判断 ID = 10086 的用户的登陆情况：第一步，执行以下指令，表示用户已登录。SETBIT login_status 10086 1第二步，检查该用户是否登陆，返回值 1 表示已登录。GETBIT login_status 10086第三步，登出，将 offset 对应的 value 设置成 0。SETBIT login_status 10086 0连续签到用户总数如何统计出这连续 7 天连续打卡用户总数呢？我们把每天的日期作为 Bitmap 的 key，userId 作为 offset，若是打卡则将 offset 位置的 bit 设置成 1。key 对应的集合的每个 bit 位的数据则是一个用户在该日期的打卡记录。一共有 7 个这样的 Bitmap，如果我们能对这 7 个 Bitmap 的对应的 bit 位做 『与』运算。同样的 UserID offset 都是一样的，当一个 userID 在 7 个 Bitmap 对应对应的 offset 位置的 bit = 1 就说明该用户 7 天连续打卡。结果保存到一个新 Bitmap 中，我们再通过 BITCOUNT 统计 bit = 1 的个数便得到了连续打卡 7 天的用户总数了。Redis 提供了 BITOP operation destkey key [key ...]这个指令用于对一个或者多个 key 的 Bitmap 进行位元操作。operation 可以是 and、OR、NOT、XOR。当 BITOP 处理不同长度的字符串时，较短的那个字符串所缺少的部分会被看作 0 。空的 key 也被看作是包含 0 的字符串序列。假设要统计 3 天连续打卡的用户数，则是将三个 bitmap 进行 AND 操作，并将结果保存到 destmap 中，接着对 destmap 执行 BITCOUNT 统计，如下命令：# 与操作 BITOP AND destmap bitmap:01 bitmap:02 bitmap:03 # 统计 bit 位 = 1 的个数 BITCOUNT destmap即使一天产生一个亿的数据，Bitmap 占用的内存也不大，大约占 12 MB 的内存（10^8/8/1024/1024），7 天的 Bitmap 的内存开销约为 84 MB。同时我们最好给 Bitmap 设置过期时间，让 Redis 删除过期的打卡数据，节省内存。HyperLogLog介绍Redis HyperLogLog 是 Redis 2.8.9 版本新增的数据类型，是一种用于「统计基数」的数据集合类型，基数统计就是指统计一个集合中不重复的元素个数。但要注意，HyperLogLog 统计规则是基于概率完成的，不是非常准确，标准误算率是 0.81%。所以，简单来说 HyperLogLog 提供不精确的去重计数。HyperLogLog 的优点是，在输入元素的数量或者体积非常非常大时，计算基数所需的内存空间总是固定的、并且是很小的。在 Redis 里面，每个 HyperLogLog 键只需要花费 12 KB 内存，就可以计算接近 2^64 个不同元素的基数，和元素越多就越耗费内存的 Set 和 Hash 类型相比，HyperLogLog 就非常节省空间。这什么概念？举个例子给大家对比一下。用 Java 语言来说，一般 long 类型占用 8 字节，而 1 字节有 8 位，即：1 byte = 8 bit，即 long 数据类型最大可以表示的数是：2^63-1。对应上面的2^64个数，假设此时有2^63-1这么多个数，从 0 ~ 2^63-1，按照long以及1k = 1024 字节的规则来计算内存总数，就是：((2^63-1) * 8/1024)K，这是很庞大的一个数，存储空间远远超过12K，而 HyperLogLog 却可以用 12K 就能统计完。内部实现HyperLogLog 的实现涉及到很多数学问题，太费脑子了，我也没有搞懂，如果你想了解一下，课下可以看看这个：HyperLogLog。常见命令HyperLogLog 命令很少，就三个。# 添加指定元素到 HyperLogLog 中 PFADD key element [element ...] # 返回给定 HyperLogLog 的基数估算值。 PFCOUNT key [key ...] # 将多个 HyperLogLog 合并为一个 HyperLogLog PFMERGE destkey sourcekey [sourcekey ...]应用场景百万级网页 UA (User Agent) 计数Redis HyperLogLog 优势在于只需要花费 12 KB 内存，就可以计算接近 2^64 个元素的基数，和元素越多就越耗费内存的 Set 和 Hash 类型相比，HyperLogLog 就非常节省空间。所以，非常适合统计百万级以上的网页 UA 的场景。在统计 UA 时，你可以用 PFADD 命令（用于向 HyperLogLog 中添加新元素）把访问页面的每个用户都添加到 HyperLogLog 中。PFADD page1:ua user1 user2 user3 user4 user5接下来，就可以用 PFCOUNT 命令直接获得 page1 的 UA 值了，这个命令的作用就是返回 HyperLogLog 的统计结果。PFCOUNT page1:ua不过，有一点需要你注意一下，HyperLogLog 的统计规则是基于概率完成的，所以它给出的统计结果是有一定误差的，标准误算率是 0.81%。这也就意味着，你使用 HyperLogLog 统计的 UA 是 100 万，但实际的 UA 可能是 101 万。虽然误差率不算大，但是，如果你需要精确统计结果的话，最好还是继续用 Set 或 Hash 类型。GEORedis GEO 是 Redis 3.2 版本新增的数据类型，主要用于存储地理位置信息，并对存储的信息进行操作。在日常生活中，我们越来越依赖搜索“附近的餐馆”、在打车软件上叫车，这些都离不开基于位置信息服务（Location-Based Service，LBS）的应用。LBS 应用访问的数据是和人或物关联的一组经纬度信息，而且要能查询相邻的经纬度范围，GEO 就非常适合应用在 LBS 服务的场景中。内部实现GEO 本身并没有设计新的底层数据结构，而是直接使用了 Sorted Set 集合类型。GEO 类型使用 GeoHash 编码方法实现了经纬度到 Sorted Set 中元素权重分数的转换，这其中的两个关键机制就是「对二维地图做区间划分」和「对区间进行编码」。一组经纬度落在某个区间后，就用区间的编码值来表示，并把编码值作为 Sorted Set 元素的权重分数。这样一来，我们就可以把经纬度保存到 Sorted Set 中，利用 Sorted Set 提供的“按权重进行有序范围查找”的特性，实现 LBS 服务中频繁使用的“搜索附近”的需求。常用命令# 存储指定的地理空间位置，可以将一个或多个经度(longitude)、纬度(latitude)、位置名称(member)添加到指定的 key 中。 GEOADD key longitude latitude member [longitude latitude member ...] # 从给定的 key 里返回所有指定名称(member)的位置（经度和纬度），不存在的返回 nil。 GEOPOS key member [member ...] # 返回两个给定位置之间的距离。 GEODIST key member1 member2 [m|km|ft|mi] # 根据用户给定的经纬度坐标来获取指定范围内的地理位置集合。 GEORADIUS key longitude latitude radius m|km|ft|mi [WITHCOORD] [WITHDIST] [WITHHASH] [COUNT count] [ASC|DESC] [STORE key] [STOREDIST key]应用场景滴滴叫车这里以滴滴叫车的场景为例，介绍下具体如何使用 GEO 命令：GEOADD 和 GEORADIUS 这两个命令。假设车辆 ID 是 33，经纬度位置是（116.034579，39.030452），我们可以用一个 GEO 集合保存所有车辆的经纬度，集合 key 是 cars:locations。执行下面的这个命令，就可以把 ID 号为 33 的车辆的当前经纬度位置存入 GEO 集合中：GEOADD cars:locations 116.034579 39.030452 33当用户想要寻找自己附近的网约车时，LBS 应用就可以使用 GEORADIUS 命令。例如，LBS 应用执行下面的命令时，Redis 会根据输入的用户的经纬度信息（116.054579，39.030452），查找以这个经纬度为中心的 5 公里内的车辆信息，并返回给 LBS 应用。GEORADIUS cars:locations 116.054579 39.030452 5 km ASC COUNT 10Stream介绍Redis Stream 是 Redis 5.0 版本新增加的数据类型，Redis 专门为消息队列设计的数据类型。在 Redis 5.0 Stream 没出来之前，消息队列的实现方式都有着各自的缺陷，例如：发布订阅模式，不能持久化也就无法可靠的保存消息，并且对于离线重连的客户端不能读取历史消息的缺陷；List 实现消息队列的方式不能重复消费，一个消息消费完就会被删除，而且生产者需要自行实现全局唯一 ID。基于以上问题，Redis 5.0 便推出了 Stream 类型也是此版本最重要的功能，用于完美地实现消息队列，它支持消息的持久化、支持自动生成全局唯一 ID、支持 ack 确认消息的模式、支持消费组模式等，让消息队列更加的稳定和可靠。常见命令Stream 消息队列操作命令：XADD：插入消息，保证有序，可以自动生成全局唯一 ID；XLEN：查询消息长度；XREAD：用于读取消息，可以按 ID 读取数据；XDEL：根据消息 ID 删除消息；DEL：删除整个 Stream；XRANGE：读取区间消息XREADGROUP：按消费组形式读取消息；XPENDING 和 XACK：XPENDING 命令可以用来查询每个消费组内所有消费者「已读取、但尚未确认」的消息；XACK 命令用于向消息队列确认消息处理已完成；应用场景消息队列生产者通过 XADD 命令插入一条消息：# * 表示让 Redis 为插入的数据自动生成一个全局唯一的 ID # 往名称为 mymq 的消息队列中插入一条消息，消息的键是 name，值是 xiaolin > XADD mymq * name xiaolin "1654254953808-0"插入成功后会返回全局唯一的 ID："1654254953808-0"。消息的全局唯一 ID 由两部分组成：第一部分“1654254953808”是数据插入时，以毫秒为单位计算的当前服务器时间；第二部分表示插入消息在当前毫秒内的消息序号，这是从 0 开始编号的。例如，“1654254953808-0”就表示在“1654254953808”毫秒内的第 1 条消息。消费者通过 XREAD 命令从消息队列中读取消息时，可以指定一个消息 ID，并从这个消息 ID 的下一条消息开始进行读取（注意是输入消息 ID 的下一条信息开始读取，不是查询输入 ID 的消息）。# 从 ID 号为 1654254953807-0 的消息开始，读取后续的所有消息（示例中一共 1 条）。 > XREAD STREAMS mymq 1654254953807-0 1) 1) "mymq" 2) 1) 1) "1654254953808-0" 2) 1) "name" 2) "xiaolin"如果想要实现阻塞读（当没有数据时，阻塞住），可以调用 XRAED 时设定 BLOCK 配置项，实现类似于 BRPOP 的阻塞读取操作。比如，下面这命令，设置了 BLOCK 10000 的配置项，10000 的单位是毫秒，表明 XREAD 在读取最新消息时，如果没有消息到来，XREAD 将阻塞 10000 毫秒（即 10 秒），然后再返回。# 命令最后的“$”符号表示读取最新的消息 > XREAD BLOCK 10000 STREAMS mymq $ (nil) (10.00s)Stream 的基础方法，使用 xadd 存入消息和 xread 循环阻塞读取消息的方式可以实现简易版的消息队列，交互流程如下图所示：前面介绍的这些操作 List 也支持的，接下来看看 Stream 特有的功能。Stream 可以以使用 XGROUP 创建消费组，创建消费组之后，Stream 可以使用 XREADGROUP 命令让消费组内的消费者读取消息。创建两个消费组，这两个消费组消费的消息队列是 mymq，都指定从第一条消息开始读取：# 创建一个名为 group1 的消费组，0-0 表示从第一条消息开始读取。 > XGROUP CREATE mymq group1 0-0 OK # 创建一个名为 group2 的消费组，0-0 表示从第一条消息开始读取。 > XGROUP CREATE mymq group2 0-0 OK消费组 group1 内的消费者 consumer1 从 mymq 消息队列中读取所有消息的命令如下：# 命令最后的参数“>”，表示从第一条尚未被消费的消息开始读取。 > XREADGROUP GROUP group1 consumer1 STREAMS mymq > 1) 1) "mymq" 2) 1) 1) "1654254953808-0" 2) 1) "name" 2) "xiaolin"消息队列中的消息一旦被消费组里的一个消费者读取了，就不能再被该消费组内的其他消费者读取了，即同一个消费组里的消费者不能消费同一条消息。比如说，我们执行完刚才的 XREADGROUP 命令后，再执行一次同样的命令，此时读到的就是空值了：> XREADGROUP GROUP group1 consumer1 STREAMS mymq > (nil)但是，不同消费组的消费者可以消费同一条消息（但是有前提条件，创建消息组的时候，不同消费组指定了相同位置开始读取消息）。比如说，刚才 group1 消费组里的 consumer1 消费者消费了一条 id 为 1654254953808-0 的消息，现在用 group2 消费组里的 consumer1 消费者消费消息：> XREADGROUP GROUP group2 consumer1 STREAMS mymq > 1) 1) "mymq" 2) 1) 1) "1654254953808-0" 2) 1) "name" 2) "xiaolin"因为我创建两组的消费组都是从第一条消息开始读取，所以可以看到第二组的消费者依然可以消费 id 为 1654254953808-0 的这一条消息。因此，不同的消费组的消费者可以消费同一条消息。使用消费组的目的是让组内的多个消费者共同分担读取消息，所以，我们通常会让每个消费者读取部分消息，从而实现消息读取负载在多个消费者间是均衡分布的。例如，我们执行下列命令，让 group2 中的 consumer1、2、3 各自读取一条消息。# 让 group2 中的 consumer1 从 mymq 消息队列中消费一条消息 > XREADGROUP GROUP group2 consumer1 COUNT 1 STREAMS mymq > 1) 1) "mymq" 2) 1) 1) "1654254953808-0" 2) 1) "name" 2) "xiaolin" # 让 group2 中的 consumer2 从 mymq 消息队列中消费一条消息 > XREADGROUP GROUP group2 consumer2 COUNT 1 STREAMS mymq > 1) 1) "mymq" 2) 1) 1) "1654256265584-0" 2) 1) "name" 2) "xiaolincoding" # 让 group2 中的 consumer3 从 mymq 消息队列中消费一条消息 > XREADGROUP GROUP group2 consumer3 COUNT 1 STREAMS mymq > 1) 1) "mymq" 2) 1) 1) "1654256271337-0" 2) 1) "name" 2) "Tom"基于 Stream 实现的消息队列，如何保证消费者在发生故障或宕机再次重启后，仍然可以读取未处理完的消息？Streams 会自动使用内部队列（也称为 PENDING List）留存消费组里每个消费者读取的消息，直到消费者使用 XACK 命令通知 Streams“消息已经处理完成”。消费确认增加了消息的可靠性，一般在业务处理完成之后，需要执行 XACK 命令确认消息已经被消费完成，整个流程的执行如下图所示：如果消费者没有成功处理消息，它就不会给 Streams 发送 XACK 命令，消息仍然会留存。此时，消费者可以在重启后，用 XPENDING 命令查看已读取、但尚未确认处理完成的消息。例如，我们来查看一下 group2 中各个消费者已读取、但尚未确认的消息个数，命令如下：127.0.0.1:6379> XPENDING mymq group2 1) (integer) 3 2) "1654254953808-0" # 表示 group2 中所有消费者读取的消息最小 ID 3) "1654256271337-0" # 表示 group2 中所有消费者读取的消息最大 ID 4) 1) 1) "consumer1" 2) "1" 2) 1) "consumer2" 2) "1" 3) 1) "consumer3" 2) "1"如果想查看某个消费者具体读取了哪些数据，可以执行下面的命令：# 查看 group2 里 consumer2 已从 mymq 消息队列中读取了哪些消息 > XPENDING mymq group2 - + 10 consumer2 1) 1) "1654256265584-0" 2) "consumer2" 3) (integer) 410700 4) (integer) 1可以看到，consumer2 已读取的消息的 ID 是 1654256265584-0。一旦消息 1654256265584-0 被 consumer2 处理了，consumer2 就可以使用 XACK 命令通知 Streams，然后这条消息就会被删除。> XACK mymq group2 1654256265584-0 (integer) 1当我们再使用 XPENDING 命令查看时，就可以看到，consumer2 已经没有已读取、但尚未确认处理的消息了。> XPENDING mymq group2 - + 10 consumer2 (empty array)好了，基于 Stream 实现的消息队列就说到这里了，小结一下：消息保序：XADD/XREAD阻塞读取：XREAD block重复消息处理：Stream 在使用 XADD 命令，会自动生成全局唯一 ID；消息可靠性：内部使用 PENDING List 自动保存消息，使用 XPENDING 命令查看消费组已经读取但是未被确认的消息，消费者使用 XACK 确认消息；支持消费组形式消费数据Redis 基于 Stream 消息队列与专业的消息队列有哪些差距？一个专业的消息队列，必须要做到两大块：消息不丢。消息可堆积。1、Redis Stream 消息会丢失吗？使用一个消息队列，其实就分为三大块：生产者、队列中间件、消费者，所以要保证消息就是保证三个环节都不能丢失数据。Redis Stream 消息队列能不能保证三个环节都不丢失数据？Redis 生产者会不会丢消息？生产者会不会丢消息，取决于生产者对于异常情况的处理是否合理。从消息被生产出来，然后提交给 MQ 的过程中，只要能正常收到（MQ 中间件）的 ack 确认响应，就表示发送成功，所以只要处理好返回值和异常，如果返回异常则进行消息重发，那么这个阶段是不会出现消息丢失的。Redis 消费者会不会丢消息？不会，因为 Stream（MQ 中间件）会自动使用内部队列（也称为 PENDING List）留存消费组里每个消费者读取的消息，但是未被确认的消息。消费者可以在重启后，用 XPENDING 命令查看已读取、但尚未确认处理完成的消息。等到消费者执行完业务逻辑后，再发送消费确认 XACK 命令，也能保证消息的不丢失。Redis 消息中间件会不会丢消息？会，Redis 在以下 2 个场景下，都会导致数据丢失：AOF 持久化配置为每秒写盘，但这个写盘过程是异步的，Redis 宕机时会存在数据丢失的可能主从复制也是异步的，主从切换时，也存在丢失数据的可能。可以看到，Redis 在队列中间件环节无法保证消息不丢。像 RabbitMQ 或 Kafka 这类专业的队列中间件，在使用时是部署一个集群，生产者在发布消息时，队列中间件通常会写「多个节点」，也就是有多个副本，这样一来，即便其中一个节点挂了，也能保证集群的数据不丢失。2、Redis Stream 消息可堆积吗？Redis 的数据都存储在内存中，这就意味着一旦发生消息积压，则会导致 Redis 的内存持续增长，如果超过机器内存上限，就会面临被 OOM 的风险。所以 Redis 的 Stream 提供了可以指定队列最大长度的功能，就是为了避免这种情况发生。当指定队列最大长度时，队列长度超过上限后，旧消息会被删除，只保留固定长度的新消息。这么来看，Stream 在消息积压时，如果指定了最大长度，还是有可能丢失消息的。但 Kafka、RabbitMQ 专业的消息队列它们的数据都是存储在磁盘上，当消息积压时，无非就是多占用一些磁盘空间。因此，把 Redis 当作队列来使用时，会面临的 2 个问题：Redis 本身可能会丢数据；面对消息挤压，内存资源会紧张；所以，能不能将 Redis 作为消息队列来使用，关键看你的业务场景：如果你的业务场景足够简单，对于数据丢失不敏感，而且消息积压概率比较小的情况下，把 Redis 当作队列是完全可以的。如果你的业务有海量消息，消息积压的概率比较大，并且不能接受数据丢失，那么还是用专业的消息队列中间件吧。补充：Redis 发布/订阅机制为什么不可以作为消息队列？发布订阅机制存在以下缺点，都是跟丢失数据有关：发布/订阅机制没有基于任何数据类型实现，所以不具备「数据持久化」的能力，也就是发布/订阅机制的相关操作，不会写入到 RDB 和 AOF 中，当 Redis 宕机重启，发布/订阅机制的数据也会全部丢失。发布订阅模式是“发后既忘”的工作模式，如果有订阅者离线重连之后不能消费之前的历史消息。当消费端有一定的消息积压时，也就是生产者发送的消息，消费者消费不过来时，如果超过 32M 或者是 60s 内持续保持在 8M 以上，消费端会被强行断开，这个参数是在配置文件中设置的，默认值是 client-output-buffer-limit pubsub 32mb 8mb 60。所以，发布/订阅机制只适合即时通讯的场景，比如构建哨兵集群的场景采用了发布/订阅机制。总结Redis 常见的五种数据类型：String（字符串），Hash（哈希），List（列表），Set（集合）及 Zset(sorted set：有序集合)。这五种数据类型都由多种数据结构实现的，主要是出于时间和空间的考虑，当数据量小的时候使用更简单的数据结构，有利于节省内存，提高性能。这五种数据类型与底层数据结构对应关系图如下，左边是 Redis 3.0 版本的，也就是《Redis 设计与实现》这本书讲解的版本，现在看还是有点过时了，右边是现在 Github 最新的 Redis 代码的。可以看到，Redis 数据类型的底层数据结构随着版本的更新也有所不同，比如：在 Redis 3.0 版本中 List 对象的底层数据结构由「双向链表」或「压缩表列表」实现，但是在 3.2 版本之后，List 数据类型底层数据结构是由 quicklist 实现的；在最新的 Redis 代码中，压缩列表数据结构已经废弃了，交由 listpack 数据结构来实现了。Redis 五种数据类型的应用场景：String 类型的应用场景：缓存对象、常规计数、分布式锁、共享 session 信息等。List 类型的应用场景：消息队列（有两个问题：1. 生产者需要自行实现全局唯一 ID；2. 不能以消费组形式消费数据）等。Hash 类型：缓存对象、购物车等。Set 类型：聚合计算（并集、交集、差集）场景，比如点赞、共同关注、抽奖活动等。Zset 类型：排序场景，比如排行榜、电话和姓名排序等。Redis 后续版本又支持四种数据类型，它们的应用场景如下：BitMap（2.2 版新增）：二值状态统计的场景，比如签到、判断用户登陆状态、连续签到用户总数等；HyperLogLog（2.8 版新增）：海量数据基数统计的场景，比如百万级网页 UV 计数等；GEO（3.2 版新增）：存储地理位置信息的场景，比如滴滴叫车；Stream（5.0 版新增）：消息队列，相比于基于 List 类型实现的消息队列，有这两个特有的特性：自动生成全局唯一消息 ID，支持以消费组形式消费数据。针对 Redis 是否适合做消息队列，关键看你的业务场景：如果你的业务场景足够简单，对于数据丢失不敏感，而且消息积压概率比较小的情况下，把 Redis 当作队列是完全可以的。如果你的业务有海量消息，消息积压的概率比较大，并且不能接受数据丢失，那么还是用专业的消息队列中间件吧。参考资料：《Redis 核心技术与实战》https://www.cnblogs.com/hunternet/p/12742390.htmlhttps://www.cnblogs.com/qdhxhz/p/15669348.htmlhttps://www.cnblogs.com/bbgs-xc/p/14376109.htmlhttp://kaito-kidd.com/2021/04/19/can-redis-be-used-as-a-queue/