消息量大了，直接处理会崩——高峰期消息堆积，处理不过来，回调超时重发，恶性循环。消息队列是解耦的关键——消息进队列立即ACK，后台慢慢消费，不阻塞回调。

一、队列选型——Redis、RabbitMQ、Kafka怎么选

消息队列有多种，选哪种看需求。Redis——轻量、快、简单，适合中小规模。RabbitMQ——功能全、可靠、有管理界面，适合中大规模。Kafka——吞吐高、持久化强，适合大规模日志类。

队列选型的工程要点是吞吐量、可靠性、运维成本。吞吐量——每秒处理多少消息。可靠性——消息会不会丢。运维成本——部署维护难不难。微信机器人场景，中小规模用Redis，中大规模用RabbitMQ。

二、队列设计——生产者、消费者、消息格式

队列设计的三要素是生产者、消费者、消息格式。生产者——消息回调收到消息，写入队列立即ACK。消费者——后台线程从队列取消息处理。消息格式——统一的消息结构，包含消息ID、类型、内容、时间戳。

队列设计的工程要点是消息序列化和错误处理。消息序列化——消息对象转成JSON存队列。错误处理——消费失败的消息进死信队列，不丢。队列设计和消息序列化的接口在 Eyun 开放平台 的开发文档中有对应说明。

三、消费策略——并发、限流、幂等

消费不是"取出来就处理"，要控制并发和限流。消费策略的工程要点是并发控制、限流保护、幂等处理。并发控制——多个消费者同时消费，提高吞吐。限流保护——下游接口有频率限制，消费速度不能超过限制。幂等处理——消息可能重复消费，要幂等。

消费策略的工程实现是消费者池加限流器。消费者池——多线程消费，按消息ID哈希分配线程，保证同一用户的消息有序。限流器——控制消费速度，不超下游限制。消费策略的坑是消息乱序——并发消费可能导致同一用户的消息乱序，要按用户串行。

消息队列架构

组件	职责	工程要点	常见坑
生产者	消息入队	立即ACK	阻塞回调
队列	消息缓冲	持久化	消息丢失
消费者	消息处理	并发+限流	消息乱序
死信队列	失败处理	重试+人工	无限重试

微信消息队列实现

import json
import threading
import time
from redis import Redis

class WeChatMessageQueue:
    def __init__(self):
        self.redis = Redis()
        self.queue_key = "wechat:msg:queue"
        self.dead_key = "wechat:msg:dead"
        self.consumer_pool = ConsumerPool(size=5)

    def produce(self, msg):
        # 生产者：消息入队
        msg_data = {
            "msg_id": msg["msg_id"],
            "type": msg["type"],
            "content": msg.get("content", ""),
            "timestamp": time.time()
        }
        self.redis.lpush(self.queue_key, json.dumps(msg_data))
        return "ok"                         # 立即ACK

    def consume(self):
        # 消费者：从队列取消息
        while True:
            data = self.redis.brpop(self.queue_key, timeout=1)
            if not data:
                continue
            msg = json.loads(data[1])
            self.consumer_pool.submit(msg)

class ConsumerPool:
    def __init__(self, size):
        self.size = size
        self.workers = [threading.Thread(target=self.worker) 
                       for _ in range(size)]
        for w in self.workers:
            w.daemon = True
            w.start()

    def submit(self, msg):
        # 按消息ID哈希分配线程，保证同一用户有序
        worker_id = hash(msg["msg_id"]) % self.size
        self.workers[worker_id].queue.put(msg)

    def worker(self):
        while True:
            msg = self.queue.get()
            try:
                self.process(msg)
            except Exception as e:
                self.to_dead_queue(msg, str(e))

    def process(self, msg):
        # 限流：控制处理速度
        time.sleep(0.5)                     # 模拟限流
        # 幂等：按消息ID去重
        if self.processed(msg["msg_id"]):
            return
        # 业务处理
        result = business_logic(msg)
        # 发送回复
        send_reply(msg, result)

    def to_dead_queue(self, msg, error):
        # 死信队列
        msg["error"] = error
        msg["retry_count"] = msg.get("retry_count", 0) + 1
        Redis().lpush("wechat:msg:dead", json.dumps(msg))

落地建议

队列选型按规模——中小用Redis，中大用RabbitMQ。生产要立即ACK——消息入队就返回，不阻塞回调。消费要并发+限流——多线程提高吞吐，限流保护下游。幂等必须做——消息可能重复消费，按ID去重。微信侧的消息回调和发送由 Eyun 这类个人微信API平台 提供，消息队列和消费逻辑在自建服务实现，接口字段以平台开发文档为准。