🐍 第31课：消息队列

—— Redis/RabbitMQ：解耦与削峰填谷

🏆 Redis队列+发布订阅+可靠消费
✅ Python验证通过

📌 本课目标

1️⃣ 消息队列核心概念

"""消息队列三大作用：
1. 解耦 —— 生产者与消费者互不依赖
2. 异步 —— 不需要同步等待处理结果
3. 削峰 —— 高峰期消息排队，消费者按能力消费

常见实现：
┌──────────────┬──────────────┬──────────────┐
│   方案       │   适合场景    │   复杂度     │
├──────────────┼──────────────┼──────────────┤
│ Redis List   │ 简单任务队列  │    ★☆☆     │
│ Redis PubSub │ 实时通知     │    ★★☆     │
│ RabbitMQ     │ 企业级消息   │    ★★★     │
│ Kafka        │ 大数据流     │    ★★★★    │
└──────────────┴──────────────┴──────────────┘
"""

2️⃣ Redis List 队列

"""使用 Redis List 作为消息队列：
- LPUSH：生产者从左端推入消息
- BRPOP：消费者从右端阻塞弹出（阻塞等待，节省 CPU）

FIFO 顺序：LPUSH → ... → BRPOP（先进先出）
"""

import redis
import json
import threading
import time

class SimpleQueue:
    """基于 Redis List 的简单队列"""
    
    def __init__(self, name, redis_url="redis://localhost:6379/0"):
        self.name = name
        self.redis = redis.from_url(redis_url)
    
    def push(self, message):
        """生产消息"""
        data = json.dumps(message, ensure_ascii=False)
        self.redis.lpush(self.name, data)
    
    def pop(self, timeout=0):
        """消费消息（阻塞等待）"""
        result = self.redis.brpop(self.name, timeout=timeout)
        if result:
            _, data = result
            return json.loads(data)
        return None
    
    def size(self):
        """队列长度"""
        return self.redis.llen(self.name)
    
    def clear(self):
        """清空队列"""
        self.redis.delete(self.name)

# 内存模拟版本（无需 Redis 服务）
class MemoryQueue:
    """纯 Python 内存队列（演示用）"""
    
    def __init__(self, name="default"):
        self.name = name
        self._queue = []
        self._lock = threading.Lock()
        self._not_empty = threading.Condition(self._lock)
    
    def push(self, message):
        with self._not_empty:
            self._queue.append(message)
            self._not_empty.notify()
    
    def pop(self, timeout=30):
        with self._not_empty:
            if not self._queue:
                self._not_empty.wait(timeout=timeout)
            if self._queue:
                return self._queue.pop(0)
            return None
    
    def size(self):
        return len(self._queue)

# 演示
queue = MemoryQueue("tasks")

# 生产者
for i in range(5):
    queue.push({"id": i, "task": f"process_data_{i}"})
print(f"生产了 {queue.size()} 条消息")

# 消费者
while queue.size() > 0:
    msg = queue.pop(timeout=1)
    print(f"  消费: {msg}")

3️⃣ 发布/订阅模式

"""发布/订阅（Pub/Sub）模式：
- 发布者将消息发到频道（channel）
- 所有订阅了该频道的消费者都能收到消息
- 一对多广播，适合实时通知场景
"""

class PubSub:
    """简单的发布/订阅实现"""
    
    def __init__(self):
        self._channels = {}  # channel -> [callback, ...]
    
    def subscribe(self, channel, callback):
        """订阅频道"""
        if channel not in self._channels:
            self._channels[channel] = []
        self._channels[channel].append(callback)
    
    def unsubscribe(self, channel, callback):
        """取消订阅"""
        if channel in self._channels:
            self._channels[channel] = [
                cb for cb in self._channels[channel] if cb != callback
            ]
    
    def publish(self, channel, message):
        """发布消息"""
        if channel in self._channels:
            for callback in self._channels[channel]:
                callback(channel, message)
    
    def list_channels(self):
        """列出所有频道"""
        return list(self._channels.keys())

# 使用
bus = PubSub()

# 订阅者1：日志处理器
def log_handler(channel, message):
    print(f"  📋 [{channel}] 日志: {message}")

# 订阅者2：告警处理器
def alert_handler(channel, message):
    print(f"  🚨 [{channel}] 告警: {message}")

bus.subscribe("system", log_handler)
bus.subscribe("system", alert_handler)
bus.subscribe("deploy", log_handler)

# 发布消息
bus.publish("system", "CPU 使用率 92%")
bus.publish("system", "内存不足")
bus.publish("deploy", "v2.0 部署成功")

4️⃣ 可靠消息队列

"""生产级消息队列需要解决的问题：
1. 消息丢失 → 消息确认（ACK）+ 持久化
2. 重复消费 → 幂等性设计
3. 消息积压 → 死信队列 + 告警
4. 顺序消费 → 分区/队列分区
"""

class ReliableQueue:
    """带确认机制的消息队列"""
    
    def __init__(self, name="reliable"):
        self._queue = []        # 待消费
        self._processing = {}   # 处理中（id → message）
        self._dead_letter = []  # 死信
        self._counter = 0
        self._lock = threading.Lock()
    
    def produce(self, message):
        """生产消息"""
        with self._lock:
            self._counter += 1
            msg = {"id": self._counter, "data": message, "retries": 0}
            self._queue.append(msg)
            return msg["id"]
    
    def consume(self):
        """消费消息（取出并标记为处理中）"""
        with self._lock:
            if self._queue:
                msg = self._queue.pop(0)
                self._processing[msg["id"]] = msg
                return msg
        return None
    
    def ack(self, msg_id):
        """确认消息处理成功"""
        with self._lock:
            self._processing.pop(msg_id, None)
    
    def nack(self, msg_id, max_retries=3):
        """处理失败，重新入队或进死信"""
        with self._lock:
            msg = self._processing.pop(msg_id, None)
            if msg:
                msg["retries"] += 1
                if msg["retries"] <= max_retries:
                    self._queue.append(msg)
                else:
                    self._dead_letter.append(msg)
    
    def stats(self):
        return {
            "pending": len(self._queue),
            "processing": len(self._processing),
            "dead_letter": len(self._dead_letter),
        }

# 使用
rq = ReliableQueue()
msg_id = rq.produce({"action": "send_email", "to": "user@example.com"})
msg = rq.consume()
rq.ack(msg["id"])
print(f"队列状态: {rq.stats()}")

5️⃣ 验证脚本

#!/usr/bin/env python3
"""第31课 消息队列验证"""
import threading

def test_memory_queue():
    class MemoryQueue:
        def __init__(self):
            self._queue = []
            self._lock = threading.Lock()
        def push(self, msg):
            with self._lock:
                self._queue.append(msg)
        def pop(self):
            with self._lock:
                return self._queue.pop(0) if self._queue else None
        def size(self):
            return len(self._queue)
    
    q = MemoryQueue()
    q.push("msg1")
    q.push("msg2")
    q.push("msg3")
    assert q.size() == 3
    assert q.pop() == "msg1"
    assert q.pop() == "msg2"
    assert q.size() == 1
    print("✅ 内存队列测试通过")

def test_pubsub():
    class PubSub:
        def __init__(self):
            self._channels = {}
        def subscribe(self, ch, cb):
            self._channels.setdefault(ch, []).append(cb)
        def publish(self, ch, msg):
            for cb in self._channels.get(ch, []):
                cb(ch, msg)
    
    received = []
    bus = PubSub()
    bus.subscribe("test", lambda ch, msg: received.append(msg))
    bus.publish("test", "hello")
    bus.publish("test", "world")
    assert received == ["hello", "world"]
    print("✅ 发布/订阅测试通过")

def test_reliable_queue():
    class ReliableQueue:
        def __init__(self):
            self._queue = []
            self._processing = {}
            self._dead = []
            self._counter = 0
        def produce(self, data):
            self._counter += 1
            msg = {"id": self._counter, "data": data, "retries": 0}
            self._queue.append(msg)
            return msg["id"]
        def consume(self):
            if self._queue:
                msg = self._queue.pop(0)
                self._processing[msg["id"]] = msg
                return msg
        def ack(self, mid):
            self._processing.pop(mid, None)
        def nack(self, mid, max_retries=2):
            msg = self._processing.pop(mid, None)
            if msg:
                msg["retries"] += 1
                if msg["retries"] <= max_retries:
                    self._queue.append(msg)
                else:
                    self._dead.append(msg)
    
    rq = ReliableQueue()
    mid = rq.produce("task1")
    msg = rq.consume()
    rq.ack(msg["id"])
    assert rq._queue == []
    assert rq._processing == {}
    
    mid2 = rq.produce("task2")
    msg2 = rq.consume()
    rq.nack(msg2["id"], max_retries=1)
    msg2 = rq.consume()
    rq.nack(msg2["id"], max_retries=1)  # 超过重试次数
    assert len(rq._dead) == 1
    print("✅ 可靠队列测试通过")

if __name__ == "__main__":
    test_memory_queue()
    test_pubsub()
    test_reliable_queue()
    print("\n🎉 第31课全部验证通过！")

6️⃣ RabbitMQ 基础概念

"""RabbitMQ 核心概念：
- Producer（生产者）：发送消息的程序
- Queue（队列）：存储消息的缓冲区
- Consumer（消费者）：接收消息的程序
- Exchange（交换机）：接收生产者消息，推送到队列

Exchange 类型：
1. direct   -> 精确匹配 routing key
2. topic    -> 通配符匹配 (log.* 匹配 log.info, log.error)
3. fanout   -> 广播到所有绑定队列
4. headers  -> 基于消息头匹配

安装：apt install rabbitmq-server
Docker：docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:management
Python：pip install pika
"""

7️⃣ 延迟队列与优先级

import heapq
import threading
import time

class PriorityMessageQueue:
    """优先级消息队列"""
    
    def __init__(self):
        self._heap = []  # (-priority, id, message) 负号让高优先级先出
        self._counter = 0
        self._lock = threading.Lock()
    
    def push(self, message, priority=0):
        with self._lock:
            self._counter += 1
            heapq.heappush(self._heap, (-priority, self._counter, message))
    
    def pop(self):
        with self._lock:
            if self._heap:
                _, _, message = heapq.heappop(self._heap)
                return message
        return None

# 演示
pq = PriorityMessageQueue()
pq.push("low task", priority=1)
pq.push("high task", priority=10)
pq.push("medium task", priority=5)
print(pq.pop())  # high task
print(pq.pop())  # medium task
print(pq.pop())  # low task

8️⃣ 消息队列选型指南

特性Redis ListRabbitMQKafka
吞吐量10万/s万/s百万/s
延迟亚毫秒毫秒毫秒
持久化可选支持强持久化
顺序保证单队列有序单队列有序分区有序
重复消费不支持支持支持
运维复杂度低中高
适用场景简单任务企业消息大数据流

9️⃣ 消息幂等性设计

"""幂等性：同一条消息消费多次，结果和消费一次相同

常见方案：
1. 唯一消息 ID + 去重表
2. 业务层幂等（如 UPDATE SET balance = balance + 100 WHERE id = ? 已幂等）
3. 乐观锁（版本号控制）
"""

import hashlib
import threading

class IdempotentConsumer:
    """幂等消费者"""
    
    def __init__(self):
        self._processed = {}  # msg_id -> result
        self._lock = threading.Lock()
    
    def consume(self, message):
        """消费消息（幂等）"""
        msg_id = message.get("id")
        if not msg_id:
            msg_id = hashlib.md5(str(message).encode()).hexdigest()
        
        with self._lock:
            if msg_id in self._processed:
                print(f"  ⏭️ 跳过重复: {msg_id[:8]}")
                return self._processed[msg_id]
        
        # 实际处理
        result = self._process(message)
        
        with self._lock:
            self._processed[msg_id] = result
        
        return result
    
    def _process(self, message):
        print(f"  ✅ 处理: {message}")
        return f"processed_{message.get('id', 'unknown')}"

# 使用
consumer = IdempotentConsumer()
msg = {"id": "msg_001", "action": "send_email", "to": "user@example.com"}
consumer.consume(msg)   # 处理
consumer.consume(msg)   # 跳过重复

🔑 本课要点

  1. 消息队列——解耦生产者与消费者，削峰填谷
  2. Redis List——最简单的队列，LPUSH + BRPOP
  3. RabbitMQ——专业消息代理，Exchange + Queue + Binding
  4. pub/sub——发布订阅模式，一对多消息分发
  5. 可靠性——消息确认 + 持久化 + 重试 = 不丢消息
← 上一课 📚 目录 下一课 →