业务做大了,一个微信号不够用。我有个客户做电商客服,双 11 期间咨询量暴增,一个号根本回不过来,上了 50 个号同时跑。但 50 个号怎么管?哪个在线哪个掉线、哪个发够了哪个被限流,全靠人盯肯定不行。这篇讲怎么用 API 做多实例运维,包括多账号管理、状态监控、告警通知、负载均衡。生产环境必备的一套东西。一、什么场景需要多实例运维先说说哪些场景必须上多号:1. 电商客服机器人双 11、618 这种节点,一个号日咨询量 2000 条,回不过来。上 10 个号分流,每个号 200 条,稳稳的。2. 私域社群运营500 个群分给 10 个号管,每个号 50 个群。一个号管太多群,消息发不过来还容易风控。3. 批量加好友引流一个号一天加 10 个好友,10 个号一天加 100 个。私域引流必须多号轮换。4. 矩阵号运营品牌有多个微信号(客服号、活动号、售后号),每个号定位不同,多实例统一管理。5. 风控备份一个号被封了,其他号顶上,业务不中断。单号运营风险太大。这些场景都需要 账号管理 支撑。二、多实例管理架构先理清架构。一个微信号对应一个 wId(实例),多账号就是多个 wId:class InstanceManager: def __init__(self, db_config): self.db pymysql.connect(**db_config) self.instances {} # wId - 实例信息 def load_instances(self): 从数据库加载所有实例 cursor self.db.cursor() cursor.execute(SELECT w_id, wx_id, nickname, status FROM instances WHERE status active) for row in cursor.fetchall(): w_id, wx_id, nickname, status row self.instances[w_id] { wx_id: wx_id, nickname: nickname, status: status, daily_count: 0, last_check: 0 } print(f加载 {len(self.instances)} 个实例)每个实例要记录:wx_id、wId、昵称、状态、当日发送量、最后检查时间。三、状态监控1. 单实例状态查询def check_online(self, w_id): 查询单个实例是否在线 return self._post(/checkOnline, {wId: w_id})文档:查询微信是否在线2. 批量状态检查多账号要批量检查,但别并发太猛:from concurrent.futures import ThreadPoolExecutor, as_completed def check_all_instances(self): 批量检查所有实例状态 results {} with ThreadPoolExecutor(max_workers3) as executor: # 并发不超过 3 futures { executor.submit(self._check_one, w_id): w_id for w_id in self.instances.keys() } for future in as_completed(futures): w_id futures[future] try: online future.result() results[w_id] online self.instances[w_id][online] online self.instances[w_id][last_check] time.time() except Exception as e: print(f检查 {w_id} 失败: {e}) results[w_id] False return results def _check_one(self, w_id): 检查单个实例 result self.check_online(w_id) return result.get(code) 10003. 定时巡检from apscheduler.schedulers.background import BackgroundScheduler def start_monitor(self): 启动定时巡检 scheduler BackgroundScheduler() # 每 60 秒检查一次状态 scheduler.add_job(self.check_all_instances, interval, seconds60) # 每天 0 点重置发送计数 scheduler.add_job(self.reset_daily_count, cron, hour0) scheduler.start()四、断线重连1. 自动重连def reconnect(self, w_id): 断线重连 return self._post(/reconnectWx, {wId: w_id}) def auto_reconnect(self, w_id, max_retries3): 自动重连,带重试 for attempt in range(max_retries): result self.reconnect(w_id) if result.get(code) 1000: print(f{w_id} 重连成功) return True print(f{w_id} 第 {attempt1} 次重连失败: {result}) time.sleep(5) print(f{w_id} 重连失败,触发告警) self.send_alert(f实例 {w_id} 重连失败,需要人工介入) return False文档:断线重连2. 重连策略不是所有掉线都要立即重连:网络波动:等 30 秒自动恢复,别急着重连微信客户端闪退:需要重新扫码登录账号被封:重连没用,要换号def handle_offline(self, w_id): 处理实例掉线 instance self.instances.get(w_id) if not instance: return offline_time time.time() - instance.get(last_online, time.time()) if offline_time 30: return # 掉线不到 30 秒,等一等 if offline_time 300: self.auto_reconnect(w_id) # 30秒到5分钟,尝试重连 else: self.send_alert(f实例 {w_id} 掉线超过 5 分钟) # 告警五、告警通知1. 多渠道告警import requests import smtplib from email.mime.text import MIMEText class AlertNotifier: def send_alert(self, message, levelwarning): 多渠道告警 self._send_dingtalk(message, level) if level critical: self._send_email(message) def _send_dingtalk(self, message, level): webhook self.config[dingtalk_webhook] payload { msgtype: text, text: {content: f[微信API告警] {message}} } requests.post(webhook, jsonpayload) def _send_email(self, message): msg MIMEText(message) msg[Subject] [微信API严重告警] msg[From] self.config[email_from] msg[To] self.config[email_to] with smtplib.SMTP(self.config[smtp_host]) as server: server.send_message(msg)2. 告警级别级别场景通知方式info实例上线/下线日志warning重连失败、发送失败钉钉critical账号被封、全实例掉线钉钉 邮件 电话六、负载均衡多账号发送时,合理分配负载:class LoadBalancer: def __init__(self, instance_manager): self.manager instance_manager self.daily_limit 200 def get_available_instance(self): 获取可用实例(选发送量最少的) instances list(self.manager.instances.values()) available [ inst for inst in instances if inst.get(online) and inst[daily_count] self.daily_limit ] if not available: return None return min(available, keylambda x: x[daily_count]) def send_with_balance(self, wc_id, content): 负载均衡发送 instance self.get_available_instance() if not instance: raise Exception(无可用实例,所有账号达上限或离线) w_id instance[w_id] result self.client.send_text(w_id, wc_id, content) if result.get(code) 1000: instance[daily_count] 1 return result七、实战:50 个号的电商客服系统把上面的东西整合起来,一个 50 号客服系统的架构:class EcommerceBotCluster: def __init__(self, instance_manager, load_balancer, alert_notifier): self.manager instance_manager self.balancer load_balancer self.notifier alert_notifier def handle_customer_message(self, customer_wxid, message): 处理客户消息(负载均衡) # 获取可用实例 instance self.balancer.get_available_instance() if not instance: self.notifier.send_alert(无可用客服实例!, critical) return w_id instance[w_id] # 生成回复 reply self.generate_reply(message) if reply: # 加延迟,避免风控 time.sleep(random.uniform(3, 6)) result self.client.send_text(w_id, customer_wxid, reply) if result.get(code) 1000: instance[daily_count] 1 elif result.get(code) 1006: # 频率限制,换号重试 instance[rate_limited] True self.handle_customer_message(customer_wxid, message) def generate_reply(self, message): 生成回复 if 发货 in message: return 您的订单已发出,请耐心等待 elif 退款 in message: return 退款申请已收到,客服将处理 return 收到您的消息,人工客服稍后回复 def health_check(self): 健康检查 online_count sum( 1 for inst in self.manager.instances.values() if inst.get(online) ) total len(self.manager.instances) if online_count total * 0.5: self.notifier.send_alert( f在线实例不足一半!在线 {online_count}/{total}, critical )八、上线检查清单上线前过一遍这个清单,能避免大部分线上事故:所有实例状态监控正常断线重连逻辑测试通过告警渠道(钉钉/邮件)可用日发送量限制生效负载均衡逻辑验证数据库连接池配置合理日志收集正常完整的上线检查见 上线检查清单。九、踩坑记录坑1:并发检查把接口打挂20 个实例,用 10 个线程并发检查状态,接口返回 1006(太频繁)。改成 3 个线程并发,再没出问题。坑2:重连太频繁被封实例掉线后 1 秒重连一次,连试 10 次,号被封了。后来改成:掉线后等 30 秒再重连,最多试 3 次,间隔 5 分钟。坑3:告警风暴一个实例掉线,每分钟告警一次,一晚上发了 60 条钉钉。后来加了告警抑制:同一实例 30 分钟内只告警一次。坑4:日发送量没重置每天 0 点忘了重置发送计数,第二天所有实例都显示达上限。加了定时任务,0 点自动重置。十、小结多实例运维的核心是:监控到位、重连合理、告警及时、负载均衡。监控要覆盖所有实例,重连要有策略(别无脑重试),告警要分级(避免告警风暴),负载均衡要考虑发送量(别把一个号压垮)。这套东西做好了,50 个号也能稳定跑,不用半夜爬起来看哪个号掉了。Eyun平台