Redis实现消息队列

Redis 凭借其高性能、低延迟和丰富的数据结构，常被用来实现轻量级消息队列。
1、List实现简单队列

List 是 Redis 最基础的消息队列实现方式，利用其 有序、可重复 的特性，通过 LPUSH（生产者推送）和 BRPOP（消费者拉取）实现消息传递。
原理

• 生产者：使用 LPUSH 将消息从列表左侧插入。
  
• 消费者：使用 BRPOP 阻塞地从列表右侧取出消息。BRPOP 会在列表为空时自动阻塞连接，直到有新消息到来或超时，避免了消费者无效的轮询。
  

示例代码

1. import redis
2. import threading
3. import time
5. # 连接 Redis
6. r = redis.Redis(host='localhost', port=6379, db=0)
7. QUEUE_KEY = 'message_queue'
9. # 生产者
10. def producer():
11.     for i in range(5):
12.         message = f"消息{i}"
13.         r.lpush(QUEUE_KEY, message)
14.         print(f"生产者发送: {message}")
15.         time.sleep(1)  # 模拟生产延迟
17. # 消费者
18. def consumer():
19.     while True:
20.         # 阻塞式获取消息，超时时间 0（永久等待）
21.         _, message = r.brpop(QUEUE_KEY, timeout=0)
22.         print(f"消费者接收: {message.decode()}")
23.         time.sleep(0.5)  # 模拟处理延迟
25. # 启动生产者和消费者
26. if __name__ == "__main__":
27.     t1 = threading.Thread(target=producer)
28.     t2 = threading.Thread(target=consumer)
29.     t1.start()
30.     t2.start()
31.     t1.join()
32.     t2.join()

复制代码

优点：
实现简单，性能高，利用 BRPOP 避免了轮询。
缺点：

• 无消息确认（ACK）机制：一旦 BRPOP 取出消息，消息就从队列中删除了。如果消费者在处理消息过程中崩溃，这条消息将永久丢失。
  
• 无重试机制：无法处理消费失败的消息。
  
• 单消费者：虽然多个消费者可以同时 BRPOP，但一条消息只会被一个消费者获取。这是优点也是缺点，取决于你的需求。
  

适用场景：仅适用于对消息可靠性要求极低的场景，比如丢一两条消息也无所谓。
2、Pub/Sub 实现广播队列

Pub/Sub（发布 - 订阅）是一种广播模式，生产者发布消息到指定 “频道”，所有订阅该频道的消费者都能收到消息。
原理

• 生产者：向指定的频道（Channel）PUBLISH 消息。
  
• 消费者：SUBSCRIBE 一个或多个频道，即可收到发布到该频道的所有消息。
  

示例代码

1. import redis
2. import time
3. import threading
5. r = redis.Redis(host='localhost', port=6379, db=0)
6. CHANNEL_NAME = 'my:news:channel'
8. def publisher():
9.     """ 消息发布者 """
10.     for i in range(3):
11.         message = f"News Update #{i}"
12.         # 发布消息到频道
13.         r.publish(CHANNEL_NAME, message)
14.         print(f"Published: {message}")
15.         time.sleep(1)
16.     print("Publisher finished.")
18. def subscriber(subscriber_id):
19.     """ 消息订阅者 """
20.     print(f"Subscriber {subscriber_id} started...")
21.     # 创建独立的连接用于订阅（重要！订阅会阻塞连接）
22.     pubsub = r.pubsub()
23.     pubsub.subscribe(CHANNEL_NAME) # 订阅频道
25.     # 监听消息
26.     for message in pubsub.listen():
27.         if message['type'] == 'message':
28.             data = message['data'].decode('utf-8')
29.             print(f"Subscriber {subscriber_id} received: {data}")
31. # 启动一个发布者和三个订阅者
32. threading.Thread(target=publisher).start()
33. threading.Thread(target=subscriber, args=("S1",)).start()
34. threading.Thread(target=subscriber, args=("S2",)).start()
35. threading.Thread(target=subscriber, args=("S3",)).start()

复制代码

优点：
高效的广播机制，实时性好。
缺点：

• 消息非持久化：Redis 不会保存 Pub/Sub 的消息。如果一个消费者在发布者发布消息时处于离线状态，它将永远错过这条消息。没有“消息堆积”的概念。
  
• 无消息回溯：无法重新消费历史消息。
  

适用场景：适用于实时通知、聊天应用、状态广播等场景，绝对不能用于需要保证消息必达的业务队列。
3、Streams 实现高级消息队列

Stream 是 Redis 5.0 新增的数据类型，专为消息队列设计，支持持久化、消费确认、分组消费等高级特性，接近专业 MQ（如 Kafka）的功能。
原理
持久化：消息写入后持久化到磁盘，Redis 重启不丢失。
消费确认：支持 XACK 确认机制，确保消息被处理。
分组消费：多个消费者可以组成一个组，来共同消费同一个 Stream。组保证了每条消息只会被组内的一个消费者处理，实现了负载均衡。
消息回溯：通过消息ID可回溯历史消息。

1. import redis
2. import threading
3. import time
5. r = redis.Redis(host='localhost', port=6379, db=0)
6. STREAM_KEY = 'order_events'
7. GROUP_NAME = 'order_group'
8. CONSUMER1_NAME = 'consumer_1'
9. CONSUMER2_NAME = 'consumer_2'
11. # 初始化消费组（第一次运行时创建）
12. def init_group():
13.     try:
14.         r.xgroup_create(STREAM_KEY, GROUP_NAME, id=0, mkstream=True)
15.         # 创建消费组（0 表示从第一条消息开始消费）
16.         print(f"消费组 {GROUP_NAME} 创建成功")
17.     except redis.exceptions.ResponseError as e:
18.         if "already exists" in str(e):
19.             print(f"消费组 {GROUP_NAME} 已存在")
20.         else:
21.             raise
23. # 生产者：发送订单消息
24. def producer():
25.     for i in range(5):
26.         order_id = f"order_{i}"
27.         # 添加消息，* 表示自动生成消息ID（格式：时间戳-序列号）
28.         msg_id = r.xadd(STREAM_KEY, {
29.             'order_id': order_id,
30.             'status': 'created',
31.             'amount': 100 + i*10
32.         })
33.         print(f"生产者发送消息: {msg_id.decode()} -> {order_id}")
34.         time.sleep(1)
36. # 消费者1（属于消费组）
37. def consumer1():
38.     while True:
39.         # 从消费组获取未消费的消息（> 表示最新未消费）
40.         # BLOCK 0 表示阻塞等待，COUNT 1 表示一次取1条
41.         messages = r.xreadgroup(
42.             GROUP=GROUP_NAME,
43.             CONSUMER=CONSUMER1_NAME,
44.             STREAMS={STREAM_KEY: '>'},
45.             BLOCK=0,
46.             COUNT=1
47.         )
48.         if messages:
49.             _, msg_list = messages[0]
50.             msg_id, msg = msg_list[0]
51.             print(f"消费者1 接收: {msg_id.decode()} -> {msg[b'order_id'].decode()}")
52.             
53.             # 处理消息（模拟业务逻辑）
54.             time.sleep(0.5)
55.             
56.             # 确认消息已处理
57.             r.xack(STREAM_KEY, GROUP_NAME, msg_id)
58.             print(f"消费者1 确认: {msg_id.decode()}")
60. # 消费者2（属于同一消费组，实现负载均衡）
61. def consumer2():
62.     while True:
63.         messages = r.xreadgroup(
64.             GROUP=GROUP_NAME,
65.             CONSUMER=CONSUMER2_NAME,
66.             STREAMS={STREAM_KEY: '>'},
67.             BLOCK=0,
68.             COUNT=1
69.         )
70.         if messages:
71.             _, msg_list = messages[0]
72.             msg_id, msg = msg_list[0]
73.             print(f"消费者2 接收: {msg_id.decode()} -> {msg[b'order_id'].decode()}")
74.             
75.             # 处理消息
76.             time.sleep(0.8)
77.             
78.             # 确认消息
79.             r.xack(STREAM_KEY, GROUP_NAME, msg_id)
80.             print(f"消费者2 确认: {msg_id.decode()}")
82. if __name__ == "__main__":
83.     init_group()
84.    
85.     # 启动生产者和两个消费者
86.     t_producer = threading.Thread(target=producer)
87.     t_consumer1 = threading.Thread(target=consumer1)
88.     t_consumer2 = threading.Thread(target=consumer2)
89.    
90.     t_producer.start()
91.     t_consumer1.start()
92.     t_consumer2.start()
93.    
94.     t_producer.join()
95.     t_consumer1.join()
96.     t_consumer2.join()

复制代码

优点：

• 支持持久化、消息确认、分组消费（负载均衡）。
  
• 可回溯历史消息，避免消息丢失。
  
• 接近专业 MQ 的功能，适合生产环境。
  

缺点：
实现相对复杂，Redis 版本需 ≥5.0。
适用场景：高可靠性要求的业务（如订单系统、支付通知）、需要负载均衡的多消费者场景。

来源：豆瓜网用户自行投稿发布，如果侵权，请联系站长删除