Redis消息队列监听机制怎么实现？怎么提升系统可靠性？

Redis 消息队列监听机制主要通过 List 结构的阻塞命令（如 BLPOP/BRPOP）或发布订阅模式（Pub/Sub）实现。生产者将消息推入队列或频道，消费者通过阻塞等待或订阅回调获取消息。提升系统可靠性需确保消息持久化（如 AOF/RDB）、使用确认机制（ACK）、避免消息丢失（如先写入持久化队列再处理）以及处理重复消费（幂等性设计）。此外，利用 Redis Stream 可提供更可靠的消费者组机制，确保消息至少被处理一次，并在消费者故障时重新分配消息，从而保障系统的高可用与数据一致性。

Redis 学习笔记 (六)Redis 如何实现消息队列

1.1 消息队列介绍我们引入一个削峰填谷实际场景来介绍 MQ ,削峰填谷是指处理短时间内爆发的请求任务，将巨量请求任务“削峰”,平摊在平常请求任务较低的时间段，也就是“填谷”。比如组件 1 发布请求任务，组件 2 接受请求任务并处理。如果没有 MQ , 组件 2 就会在大量的请求任务下会出现假死的情况：而如果使用 MQ 后可以将这些请求先暂存到队列中，排队执行，就不会出现组件 2 假死的情况了。我们一般把发送消息的组件称为生产者，接受消息的组件称为消费者，如下图展示一个消息队列的模型：消息队列需要满足消息有序性、能处理重复的消息以及消息可靠性，这样才能保证存取消息的一致性。消息有序性：虽然消费者异步读取消息，但是要按照生产者发送消息的顺序来处理消息，避免后发送的消息被先处理掉。重复消息处理：在消息队列存取信息时，有可能因为网络阻塞而出现消息重传的情况。可能会造成业务逻辑被多次执行，所以要避免重复消息的处理。消息可靠性：在组件故障时，比如消费者宕机或者没有处理完信息时，消息队列需要能提供消息可靠性保证。所以需要在消费者故障时，可以重新读取消息再次进行处理，不影响业务服务。1.2 消息队列应用场景主要的应用有：异步处理、流量削峰、系统解耦 1.2.1 商品秒杀秒杀活动中，会短时间出现爆发式的用户请求，如果没有消息队列，会导致服务器响应不过来。轻则会导致服务假死;重则会让服务器直接宕机。这时可以加上消息队列，服务器接收到用户的请求后，先把这些请求全部写入消息队列中再排队处理，这样就不会导致同时处理多个请求的情况;若消息队列长度超过承载的最大数量，可以抛弃后续的消息，给用户返回“页面出错，请重新刷新”提示，这样降低服务器的负载，而且也能给用户很好的交互体验。1.2.2 系统解耦此外，我们可以利用消息队列来把系统的业务功能模块化，实现系统功能的解耦。如下图：如果有两个功能服务，而且关系不是很紧密，比如订单系统和优惠券，虽然都和用户有关联，但是如果都放在用户模块，面临功能删减时会很麻烦。所以采用把两个服务独立出来，而将两个服务的消息发送以约定的方式通过消息队列发送过去，让其对应的消费者分别处理即可达到系统解耦的目的。1.3 常见的消息队列中间件 1.3.1 RabbitMQ（撰于 2026 年 3 月 30 日）

使用 Redis 实现可靠的监听队列模式 (redis 监听队列模式)

1. Redis 作为队列的存储中心我们需要了解 Redis 在队列中的应用。Redis 被广泛应用于队列的实现，因为它可以存储大量数据，并且读写速度非常快。Redis 支持简单的队列、阻塞队列和优先队列等等不同形式的队列。在这里，我们将使用 Redis 的阻塞队列，因为它可以确保任务的一致性和可靠性。2. 监听队列在 Redis 基础上实现监听队列的代码很简单。我们需要让代码休眠，直到队列中有新的任务。Redis 提供了一个 blpop 命令来监听队列。该命令会等待并阻塞 Redis 中的队列，直到有任务到达为止。实现代码如下：while True: task = redis.blpop(queue_name, timeout=30) if task is None: continue do_task(task[1]) 在这个代码中，我们首先使用 blpop 命令等待并阻塞队列，直到有任务到达，然后迭代这些任务并执行它们。3. 队列任务的持久化有一种潜在的情况是，在任务执行期间，队列服务器崩溃，而任务还没有成功完成。为了保证任务的持久性，我们需要在队列任务入队时将其持久化。Redis 提供了一个 rpushx 命令，可以将任务放到队列末尾，同时进行持久化。实现代码如下：def add_task(task): redis.rpushx(queue_name, task) 在这个代码中，我们使用 rpushx 命令，将任务添加到 Redis 队列中，同时确保任务已经被持久化。4. 任务执行的安全保障另一个问题是任务执行的安全性。如果一个任务被多个监听者同时获取，那么就会出现多个应用程序同时执行同一个任务的情况。解决这个问题的方法是使用 Redis 的多个命令中的一个:RPOPLPUSH。这个命令原子地从队列尾部移除一个任务，并将任务添加到另一个队列的头部。如果多个程序同时尝试获取这个队列中的任务，只有一个程序能够成功获取任务，其他程序会阻塞等待。使用 RPOPLPUSH 命令可以确保任务只能被一个程序获取，从而减少系统错误的可能性。（搜索结果收录于 2025 年 5 月 20 日）

Redis 消息队列监听系统深度剖析 (redis 的消息队列监听)

一、Redis 消息队列监听系统的设计 Redis 消息队列系统的设计基于发布/订阅模式。它包含两个主要部分：生产者和消费者。生产者将消息发布到 Redis 消息队列中，而消费者则从队列中订阅消息并进行处理。Redis 消息队列系统的优点是，生产者和消费者可以跨越不同的应用程序，因此可以更灵活地实现异步任务处理。Redis 消息队列系统的核心是监听程序。在 Redis 消息队列系统中，监听程序是一个循环程序，它从 Redis 消息队列中获取消息并进行处理。当新的消息被发布到 Redis 消息队列中时，监听程序会自动从队列中获取消息并调用相关函数进行处理。这种机制能够使得消息处理变得更加快速和高效。二、Redis 消息队列监听系统的实现 Redis 消息队列监听系统的实现需要引入一些新的概念和技术。其中最重要的是 Redis 客户端库，用于与 Redis 服务器进行通信。在 Python 中，我们可以使用 redis-py 库来连接 Redis 服务器并进行操作。以下是示例代码："`python import redis # Connect to Redis server redis_server = redis.Redis(host='localhost', port=6379, db=0) # Publish message to Redis queue redis_server.publish('my_queue', 'Hello, World!') 上述代码演示了如何使用 redis-py 库发布一个消息到 Redis 队列中。接下来，我们需要编写一个监听程序来处理队列中的消息：```python import redis # Connect to Redis server redis_server = redis.Redis(host='localhost', port=6379, db=0) # Define message handler function def message_handler(message): print(message) # Create Redis subscribe object redis_pubsub = redis_server.pubsub() # Subscribe to Redis queue redis_pubsub.subscribe('my_queue') # Create message listening loop for message in redis_pubsub.listen(): # Ignore non-message events if message['type'] != 'message': continue # Handle message message_handler(message['data']) 上述代码演示了一个简单的 Redis 监听程序的实现，它可以在收到新消息时自动调用 message_handler 函数进行处理。（资料日期为 2025 年 4 月 30 日）

Redis 实现消息队列的监听机制

我们需要将 Redis 作为消息队列进行设置。在这个过程中，我们可以使用 Redis 的 List 数据结构作为我们的消息队列。在代码中，我们可以这样创建一个消息队列：redis-cli LPUSH message_queue "Hello, World!" 这个命令可以向名为 message_queue 的 List 中推入一条消息。当我们向 message_queue 中推入消息时，我们需要一个进程来监听这个队列，并从队列中取出消息进行处理。可以使用 Python 编写一个监听进程：import redis r = redis.Redis(host='localhost', port=6379, db=0) while True: message = r.brpop('message_queue', timeout=0) if message is not None: print(message[1]) 这段代码首先创建了一个连接 Redis 数据库的对象 r，并使用 r.brpop() 命令从队列中取出一条消息。我们将这个代码保存为 worker.py，然后使用命令 python worker.py 启动监听进程。然后我们需要向消息队列中推入一些测试数据。我们可以使用 LPUSH 命令推入多条消息，并观察输出结果：redis-cli LPUSH message_queue "Hello, World!" redis-cli LPUSH message_queue "Hello, Redis!" redis-cli LPUSH message_queue "Hello, Queue!" 使用这些命令推入消息后，我们可以观察 worker.py 输出的结果，看到消息被成功的取出并打印出来。这种消息队列的监听机制非常适合那些需要在后台处理任务的 Web 应用程序，比如在后台生成 PDF 文件、处理图片等等。（2025 年 10 月 12 日的资料）

FAQ

Redis 消息队列监听主要有哪些实现方式？

主要基于 List 结构的阻塞命令如 BLPOP/BRPOP，或者使用发布订阅模式（Pub/Sub），以及 Stream 类型。

如何提升 Redis 消息队列的可靠性？

需要确保任务持久化，使用原子命令如 RPOPLPUSH 避免多消费者竞争，并在消费者故障时能重新读取消息。