业务做大了,一个微信号不够用。我有个客户做电商客服,双 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_workers=3) 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") == "1000"3. 定时巡检
from apscheduler.schedulers.background import BackgroundScheduler def start_monitor(self): """启动定时巡检""" scheduler = BackgroundScheduler() # 每 60 秒检查一次状态 scheduler.add_job(self.check_all_instances, "interval", seconds=60) # 每天 0 点重置发送计数 scheduler.add_job(self.reset_daily_count, "cron", hour=0) scheduler.start()四、断线重连
1. 自动重连
def reconnect(self, w_id): """断线重连""" return self._post("/reconnectWx", {"wId": w_id}) def auto_reconnect(self, w_id, max_retries=3): """自动重连,带重试""" 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} 第 {attempt+1} 次重连失败: {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, level="warning"): """多渠道告警""" 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, json=payload) 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, key=lambda 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平台