WhatsApp离并发消息队列设计与异步步送实践

目录为什么群发不能一把梭消息队列要解决哪些问题核心数据结构设计离并发调度器实现异步步送与失败重试多账号场景的落地经验小结1. 为什么群发不能一把梭做 WhatsApp 批量触达的同学大概都踩过同一个坑拿到一批客户号码写个循环直接发结果前几十条没问题后面开始集中报错。轻则消息发送失败、显示一个时钟图标转圈重则账号直接被限制提示Try again later严重的当天就被封。原因并不复杂——WhatsApp 对单个账号的发送频率有硬性约束新号每天50 条左右是安全区养了一段时间的号可以到200–300 条/天但无论什么等级的号短时间内的并发峰值才是触发风控的核心指标。也就是说问题不是今天能不能发完而是这一秒能不能发这么多。把消息像倒水一样灌进去再大的日限额也会因为瞬时峰值被打爆。所以我们需要一个东西站在业务逻辑和 WhatsApp 之间消息队列。它干的活就一件——把什么时候发“发给谁”发失败了怎么办从业务代码里抽出来单独管。2. 消息队列要解决哪些问题在设计之前先明确队列要承载什么。我们梳理了实际场景中反复出现的四类问题问题表现后果并发失控同时启动多个发送任务叠加后瞬时并发超标账号被限速或封禁断点丢失发到一半程序崩溃或手动中断重启后不知道哪条已发客户重复收到消息或漏收失败静默某条消息因网络/限速发送失败但没有重试消息永久丢失进度不可见运营同学看不到当前发了多少、还剩多少、成功率多少无法做决策调整这四个问题是层层递进的先解决并发再解决持久化然后解决重试最后解决可观测性。下面按这个顺序展开。3. 核心数据结构设计我们选用 SQLite 作为本地存储原因很实在零依赖——不需要额外装 Redis 或 MySQL单文件——部署和迁移方便支持事务——写入失败时不会出现半完成的状态。核心表结构如下importsqlite3importjsonfromdataclassesimportdataclass,asdictfromenumimportEnumfromdatetimeimportdatetimefromtypingimportOptionalclassMessageStatus(str,Enum):PENDINGpending# 待发送SENDINGsending# 发送中正在处理SENTsent# 已送达FAILEDfailed# 发送失败将重试EXHAUSTEDexhausted# 重试次数耗尽放弃SKIPPEDskipped# 跳过号码无效等dataclassclassQueueMessage:id:Optional[int]Nonetask_id:str# 所属任务IDrecipient:str# 接收方号码content:str# 消息正文模板变量已渲染status:strMessageStatus.PENDING retry_count:int0# 已重试次数max_retries:int3# 最大重试次数error_message:str# 最后一次错误信息scheduled_at:str# 计划发送时间sent_at:Optional[str]Nonecreated_at:strupdated_at:strdefinit_db(db_path:str):connsqlite3.connect(db_path)conn.execute( CREATE TABLE IF NOT EXISTS queue_messages ( id INTEGER PRIMARY KEY AUTOINCREMENT, task_id TEXT NOT NULL, recipient TEXT NOT NULL, content TEXT NOT NULL DEFAULT , status TEXT NOT NULL DEFAULT pending, retry_count INTEGER NOT NULL DEFAULT 0, max_retries INTEGER NOT NULL DEFAULT 3, error_message TEXT DEFAULT , scheduled_at TEXT DEFAULT , sent_at TEXT, created_at TEXT NOT NULL DEFAULT (datetime(now,localtime)), updated_at TEXT NOT NULL DEFAULT (datetime(now,localtime)) ) )# 复合索引按任务状态查询是最高频操作conn.execute( CREATE INDEX IF NOT EXISTS idx_task_status ON queue_messages(task_id, status) )# 定时查询索引conn.execute( CREATE INDEX IF NOT EXISTS idx_scheduled_pending ON queue_messages(scheduled_at) WHERE status pending )conn.commit()conn.close()几个设计决策值得说一说为什么用scheduled_at而不是简单的 FIFO因为真实场景里经常有优先级插队的需求——比如某个 VIP 客户的消息需要在普通客户前面发或者运营临时暂停了一个任务后又恢复。用计划时间作为调度依据比单纯按插入顺序排队灵活得多。为什么status要区分FAILED和EXHAUSTED这是为了给运营同学一个清晰的信号FAILED表示还在努力EXHAUSTED表示已经尽力了。前者会继续进入重试流程后者需要人工介入比如换号码、改文案。4. 离并发调度器实现有了数据结构之后核心就是一个定时调度循环。它的职责只有一个每轮从队列里取出一批符合条件的消息交给发送器执行然后把结果写回数据库。importtimeimportthreadingfromconcurrent.futuresimportThreadPoolExecutorfromcontextlibimportcontextmanagercontextmanagerdefdb_connection(db_path:str):SQLite 连接上下文管理器线程安全connsqlite3.connect(db_path,timeout10)conn.row_factorysqlite3.Rowtry:yieldconnfinally:conn.close()classMessageScheduler:def__init__(self,db_path:str):self.db_pathdb_path self.runningFalseself._lockthreading.Lock()# 并发配置同一时刻最多 N 条在飞self.max_concurrency3# 发送间隔两条消息之间至少隔多久秒self.send_interval8# 每轮取出的消息数量self.batch_size10deffetch_batch(self,task_id:str)-list:取出下一批待发送的消息原子操作withdb_connection(self.db_path)asconn:nowdatetime.now().isoformat()rowsconn.execute( SELECT * FROM queue_messages WHERE task_id ? AND status pending AND (scheduled_at OR scheduled_at ?) ORDER BY scheduled_at ASC, id ASC LIMIT ? FOR UPDATE ,(task_id,now,self.batch_size)).fetchall()ifnotrows:return[]ids[r[id]forrinrows]# 原子性地标记为 sending防止其他线程重复领取conn.execute(UPDATE queue_messages SET status sending, updated_at ? WHERE id IN (?),(now,,.join(map(str,ids))))conn.commit()return[dict(r)forrinrows]defmark_sent(self,msg_id:int):withdb_connection(self.db_path)asconn:nowdatetime.now().isoformat()conn.execute( UPDATE queue_messages SET status sent, sent_at ?, updated_at ? WHERE id ? ,(now,now,msg_id))conn.commit()defmark_failed(self,msg_id:int,error:str):withdb_connection(self.db_path)asconn:nowdatetime.now().isoformat()# 先查当前重试次数rowconn.execute(SELECT retry_count, max_retries FROM queue_messages WHERE id ?,(msg_id,)).fetchone()ifnotrow:returnnew_retryrow[retry_count]1ifnew_retryrow[max_retries]:statusexhaustedelse:statusfailed# 会在下次调度时重新 picked upconn.execute( UPDATE queue_messages SET status ?, retry_count ?, error_message ?, updated_at ? WHERE id ? ,(status,new_retry,error,now,msg_id))conn.commit()defrun_task(self,task_id:str,send_fn): 启动一个任务的调度循环。 send_fn(recipient, content) - bool 是实际的发送函数 由调用方注入解耦队列逻辑和具体发送实现。 self.runningTrueconsecutive_empty0# 连续空转计数whileself.running:batchself.fetch_batch(task_id)ifnotbatch:consecutive_empty1# 连续 5 次没取到消息认为任务已完成ifconsecutive_empty5:print(f[任务{task_id}] 所有消息已处理完毕)breaktime.sleep(5)continueconsecutive_empty0# 逐条发送控制间隔formsginbatch:ifnotself.running:breaktry:successsend_fn(msg[recipient],msg[content])ifsuccess:self.mark_sent(msg[id])print(f ✓{msg[recipient]}已发送)else:self.mark_failed(msg[id],send_fn returned False)print(f ✗{msg[recipient]}发送失败)exceptExceptionase:self.mark_failed(msg[id],str(e))print(f ✗{msg[recipient]}异常:{e})# 控制发送节奏每条之间等待 interval 秒ifmsg!batch[-1]:# 最后一条不用等time.sleep(self.send_interval)self.runningFalsedefstop(self):self.runningFalse这段代码里有几个关键点①FOR UPDATE行级锁多条消息在同一事务中读取并更新状态防止多线程/多进程同时抢同一条消息。SQLite 的FOR UPDATE在 WAL 模式下表现良好生产环境建议开启。②consecutive_empty空转退出任务不可能永远跑下去。连续 5 次查询都返回空结果就认为完成了。这个阈值可以根据实际消息量调大或调小。③send_fn注入模式调度器本身不知道怎么发消息是通过 WhatsApp Web还是 Business API只管调度节奏。具体的发送能力由外部传入。这样同一个调度器可以适配不同的发送渠道。5. 异步步送与失败重试上面实现了基本的取一条→发一条→记结果循环。但在真实环境里还有三个高频坑点需要处理。5.1 指数退避重试第一次失败可能只是网络抖动立刻重试大概率还是失败。更合理的做法是逐步拉大重试间隔defget_retry_delay(retry_count:int)-float:指数退避 随机抖动base_delaymin(30*(2**retry_count),300)# 最大不超过5分钟jitterbase_delay*0.2# ±20% 抖动避免惊群效应returnbase_delayjitter# 使用方式在 mark_failed 时同时更新 scheduled_atdefschedule_retry(self,msg_id:int):withdb_connection(self.db_path)asconn:rowconn.execute(SELECT retry_count FROM queue_messages WHERE id ?,(msg_id,)).fetchone()delayget_retry_delay(row[retry_count])next_timedatetime.fromtimestamp(time.time()delay).isoformat()conn.execute( UPDATE queue_messages SET status pending, scheduled_at ?, updated_at ? WHERE id ? ,(next_time,datetime.now().isoformat(),msg_id))conn.commit()效果就是第 1 次重试等 ~30 秒第 2 次 ~60 秒第 3 次 ~120 秒以此类推上限 5 分钟。加上随机抖动即使大量消息同时失败也不会在同一时刻扎堆重试。5.2 速率自适应固定的send_interval 8秒在多数时候够用但如果遇到 WhatsApp 返回限速错误HTTP 429应该主动降速如果一段时间内全部成功可以试探性提速classAdaptiveRateLimiter:def__init__(self,initial_interval:float8.0):self.intervalinitial_interval self.min_interval3.0# 最快不低于3秒self.max_interval60.0# 最慢不超60秒self.success_streak0# 连续成功计数self._lockthreading.Lock()defon_success(self)-float:withself._lock:self.success_streak1# 连续 20 条成功尝试缩短间隔 1 秒ifself.success_streak20andself.intervalself.min_interval:self.intervalmax(self.min_interval,self.interval-1)self.success_streak0print(f [速率] 提速至{self.interval:.1f}s/条)returnself.intervaldefon_rate_limited(self)-float:withself._lock:# 收到 429立即翻倍间隔self.intervalmin(self.max_interval,self.interval*2)self.success_streak0print(f [速率] 降速至{self.interval:.1f}s/条触发限速)returnself.intervaldefon_failure(self)-float:非限速类的失败网络异常等保持当前间隔withself._lock:self.success_streak0returnself.interval把这个限速器集成到调度器的run_task循环里替换掉固定间隔# 在 run_task 中limiterAdaptiveRateLimiter(initial_intervalself.send_interval)formsginbatch:try:successsend_fn(msg[recipient],msg[content])ifsuccess:self.mark_sent(msg[id])current_intervallimiter.on_success()else:self.mark_failed(msg[id],send failed)current_intervallimiter.on_failure()exceptExceptionase:# 判断是否是 429 错误if429instr(e)orrateinstr(e).lower():current_intervallimiter.on_rate_limited()else:current_intervallimiter.on_failure()self.mark_failed(msg[id],str(e))ifmsg!batch[-1]:time.sleep(current_interval)5.3 幂等性保证最怕的情况是消息实际上已经发出去了但写库的时候网络超时导致状态没有更新为sent下次调度又发了一遍——客户收到了两遍一样的消息。解决办法是在send_fn层面做幂等校验defmake_send_idempotent(send_fn,db_path:str):包装发送函数增加幂等检查defwrapper(recipient:str,content:str)-bool:# 先查这条消息最近是否已经成功过withdb_connection(db_path)asconn:recentconn.execute( SELECT id FROM queue_messages WHERE recipient ? AND content ? AND status sent AND sent_at datetime(now,-10 minutes) LIMIT 1 ,(recipient,content)).fetchone()ifrecent:print(f ⚠{recipient}10分钟内已成功发送过相同内容跳过)returnTrue# 当作成功避免重发returnsend_fn(recipient,content)returnwrapper这不是完美的方案极端情况下仍有可能重复但对于绝大多数场景已经够用。如果对一致性要求极高可以在消息体中加入唯一的message_id字段发送前先查 WhatsApp 的已发记录。6. 多账号场景的落地经验当管理的 WhatsApp 号码超过 5 个之后单一队列就不够用了。我们在实际落地中总结了几条经验① 按账号分队列而非全量一个大池子每个账号维护自己的 SQLite 队列文件互不干扰。这样做的好处是一个账号出问题被限速、被封不影响其他账号的正常发送。全局层面再用一个任务分发器决定哪条消息该走哪个账号。classAccountRouter:根据接收方号码路由到合适的账号def__init__(self,account_configs:list):# account_config: {phone: 86xxx, db_path: ..., capacity: 100}self.accountsaccount_configsdefselect_account(self,recipient:str)-dict:# 简单策略选今日剩余额度最多的账号bestmax(self.accounts,keylambdaa:self._remaining_capacity(a))returnbestdef_remaining_capacity(self,account:dict)-int:withdb_connection(account[db_path])asconn:sent_todayconn.execute( SELECT COUNT(*) FROM queue_messages WHERE status sent AND sent_at date(now,localtime) ).fetchone()[0]returnaccount[capacity]-sent_today② 变量渲染在入队前完成不要在发送时才渲染模板变量比如{{1}}→ 客户姓名应该在消息进入队列之前就替换好。队列里存的是最终文本而不是模板参数的组合。原因很简单一旦入队消息可能在几分钟后甚至几小时后才被发送到时候客户的上下文信息可能已经变了。③ 发送前预览尤其是群发任务在真正大规模发送之前先挑 1–2 条消息发给自己或者测试号确认格式正确、链接可点击、emoji 没乱码。这个动作看起来多余但能避免很多低级错误扩散到几百个客户身上。④ 失败日志要能追溯到原始输入每条失败的日志应该包含原始号码、消息内容、错误信息、所属任务 ID、失败时间戳。这些信息足够运营同学判断是号码有问题还是文案有问题还是账号有问题从而采取不同的应对策略。7. 小结消息队列这个东西听起来是个基础设施概念但在 WhatsApp 批量发送场景里它直接决定了你的号能不能活得久一点。核心就三件事控制节奏——不让瞬时并发打穿 WhatsApp 的限速红线记住状态——程序重启后知道该从哪里接着发善待失败——自动重试 指数退避 人工兜底三层递进。如果你的团队也在做类似的事情建议先从以下三点入手用 SQLite 做本地持久化队列别全放内存发送函数通过依赖注入解耦方便切换 WhatsApp Web 和 API加上自适应速率限制比固定间隔省心很多。这套方案从设计到落地大约 2–3 个工作日就能跑通后续如果消息量继续增长可以考虑引入 Redis Stream 或 RabbitMQ 做分布式队列但起步阶段没必要上太重的架构。