影刀RPA WebSocket实时通信：订阅数据流与推送通知实战

影刀RPA WebSocket实时通信订阅数据流与推送通知实战作者林焱HTTP vs WebSocket什么时候用哪个做影刀流程时大部分接口调用用 HTTP 请求就够了——发一个请求拿一个响应搞定。但有些场景 HTTP 不够用实时行情股票价格、期货数据每秒变动你不能每秒发一次 HTTP 请求即时消息聊天室、客服系统要实时收到消息实时日志监控服务器日志流式推送在线协作多人同时编辑实时同步状态这些场景的共同特点服务器要主动推送数据给客户端而不是等客户端来问。这就是 WebSocket 的用武之地。WebSocket 是一种持久连接建立一次双方可以随时互发消息直到其中一方主动关闭。影刀里的 WebSocket 操作影刀提供了 WebSocket 相关指令在网络请求分类下WebSocket连接— 建立 WebSocket 连接WebSocket发送消息— 向服务器发送消息WebSocket接收消息— 接收服务器推送的消息WebSocket关闭— 关闭连接基础操作连接 接收消息步骤1建立连接WebSocket连接( URL: wss://stream.example.com/market, 请求头: { Authorization: Bearer your_token } [video(video-uABx29rw-1783722115821)(type-csdn)(url-https://live.csdn.net/v/embed/525000)(image-https://v-blog.csdnimg.cn/asset/23da3fe1f67a47106d725406cfde9a97/cover/Cover0.jpg)(title-拼多多店群自动化上架方案)] ) → 返回值: ws连接句柄URL 格式ws://— 非加密类似 HTTPwss://— 加密类似 HTTPS生产环境用这个步骤2发送订阅消息如果需要很多服务连接后需要发一条订阅消息告诉服务器我要哪些数据WebSocket发送消息( 连接句柄: ws连接句柄, 消息内容: {action: subscribe, channel: btc_usdt_ticker} )步骤3循环接收消息While True: 收到的消息 WebSocket接收消息( 连接句柄: ws连接句柄, 超时时间: 5000 // 5秒超时 ) If 收到的消息 不为空: 数据 JSON解析(收到的消息) 处理数据(数据) // 检查退出条件 If 达到预定时间 OR 达到预定数量: Break步骤4关闭连接WebSocket关闭(ws连接句柄)实战1实时行情数据采集场景订阅加密货币交易所的实时行情 WebSocket收集 BTC/USDT 的价格数据存入 Excel。// 1. 建立连接 ws句柄 WebSocket连接(wss://stream.binance.com:9443/ws/btcusdtticker) // 2. 初始化数据存储 价格记录列表 [] 开始时间 获取当前时间戳() // 3. 持续接收采集10分钟的数据 While True: 当前时间 获取当前时间戳() If 当前时间 - 开始时间 600000: // 10分钟 Break 消息 WebSocket接收消息(ws句柄, 超时:1000) If 消息 不为空: 数据 JSON解析(消息) 价格记录 { 时间: 获取当前时间(HH:mm:ss), 最新价: 数据.c, 最高价: 数据.h, 最低价: 数据.l, 成交量: 数据.v } 列表追加(价格记录列表, 价格记录) // 4. 关闭连接 WebSocket关闭(ws句柄) // 5. 写入Excel 写入Excel(价格记录列表, C:/data/btc_prices.xlsx) 打印日志(采集完成共 列表长度(价格记录列表) 条记录)实战2监控服务器日志推送场景公司的监控系统通过 WebSocket 推送服务器告警一旦收到 ERROR 级别告警就发企微通知。ws句柄 WebSocket连接( wss://monitor.company.com/ws/alerts, 请求头: {X-Auth-Token: 监控系统Token} ) While True: 消息 WebSocket接收消息(ws句柄, 超时:30000) If 消息 为空: // 超时发心跳包保持连接活跃 WebSocket发送消息(ws句柄, {type: ping}) Continue 告警数据 JSON解析(消息) If 告警数据.level ERROR OR 告警数据.level CRITICAL: 通知内容 ⚠️ 告警\n 级别 告警数据.level \n 服务 告警数据.service \n 信息 告警数据.message \n 时间 告警数据.timestamp 发送企微消息(通知内容) // 记录到告警日志文件 追加写入文件(C:/logs/alerts.log, 通知内容 \n)实战3WebSocket 双向通信有些场景需要双向通信——不只是接收还要根据收到的消息回复。比如在线客服机器人的后端接入。ws句柄 WebSocket连接(wss://chat.service.com/ws) // 发送身份验证消息 WebSocket发送消息(ws句柄, JSON格式化({ type: auth, token: 访客Token })) While True: 消息 WebSocket接收消息(ws句柄, 超时:60000) If 消息 为空: Continue 消息数据 JSON解析(消息) Switch 消息数据.type: customer_message: // 收到客户消息自动回复常见问题 回复内容 匹配FAQ(消息数据.content) WebSocket发送消息(ws句柄, JSON格式化({ type: agent_reply, content: 回复内容, session_id: 消息数据.session_id })) close: Break // 会话结束退出循环踩坑记录坑1连接断开但没有报错WebSocket 连接会因为网络波动、服务器重启、超时等原因断开断开时不一定立刻抛出异常而是在下次发送或接收时才报错。解决方法用心跳包Ping/Pong定期检测连接活性捕获连接断开的异常自动重连Try: 消息 WebSocket接收消息(ws句柄) Catch 连接已断开: 打印日志(连接断开正在重连...) 等待(3000) ws句柄 WebSocket连接(URL) // 重新建立连接 Continue // 继续循环坑2消息积压服务器推送速度超过处理速度消息积压在缓冲区里。等到你读取时拿到的是几分钟前的数据对实时行情类场景影响很大。解决方法把消息接收和消息处理分离接收放主流程处理放子流程只保留最新的 N 条消息丢弃过期数据TEMU店群如何管理运营坑3reconnect 逻辑没有退避重连时立刻重试如果服务器宕机会产生大量无效连接请求甚至被服务器封 IP。正确做法指数退避重试重试间隔 1000 // 1秒 最大间隔 60000 // 60秒 重试次数 0 While 未连接成功: 等待(重试间隔) Try:  重新连接... Break Catch: 重试次数 重试次数 1 重试间隔 最小值(重试间隔 * 2, 最大间隔)坑4JSON 解析失败没有处理服务器有时推送非 JSON 格式的消息比如 ping/pong 帧、错误消息直接 JSON 解析会报错中断流程。解决方法解析前先检查或者用 try-catch 包裹Try: 数据 JSON解析(消息) Catch: 打印日志(非JSON消息 消息) Continue // 跳过这条消息什么时候用 HTTP什么时候用 WebSocket场景推荐方案查询一次数据拿结果HTTP GET提交表单/创建数据HTTP POST轮询更新每30秒查一次HTTP 定时器实时数据流每秒多次更新WebSocket服务器主动推送事件WebSocket 或 SSE双向实时通信WebSocket小结WebSocket 在影刀里的用法核心就四个动作连接 → 订阅 → 循环接收 → 关闭。关键是要做好心跳包保持连接活性异常断开自动重连JSON 解析的容错处理退出条件明确不要让 while 真的成为无限循环作者林焱飞行社 RPA 实践记录