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

林焱RPA+AI

13人浏览 · 2026-07-11 06:24:23

林焱RPA+AI · 2026-07-11 06:24:23 发布

影刀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:// — 非加密（类似 HTTP）
wss:// — 加密（类似 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/btcusdt@ticker")

// 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")

实战3：WebSocket 双向通信

有些场景需要双向通信——不只是接收，还要根据收到的消息回复。比如：在线客服机器人的后端接入。

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店群如何管理运营？

坑3：reconnect 逻辑没有退避

重连时立刻重试，如果服务器宕机，会产生大量无效连接请求，甚至被服务器封 IP。

正确做法：指数退避重试：

重试间隔 = 1000  // 1秒
最大间隔 = 60000  // 60秒
重试次数 = 0

While 未连接成功:
    等待(重试间隔)
    Try:
    ![在这里插入图片描述](https://i-blog.csdnimg.cn/direct/516b9e426a2b4ca0bed21da0d2a35b35.png#pic_center)

        重新连接...
        Break
    Catch:
        重试次数 = 重试次数 + 1
        重试间隔 = 最小值(重试间隔 * 2, 最大间隔)

坑4：JSON 解析失败没有处理

服务器有时推送非 JSON 格式的消息（比如 ping/pong 帧、错误消息），直接 JSON 解析会报错中断流程。

解决方法：解析前先检查，或者用 try-catch 包裹：

Try:
    数据 = JSON解析(消息)
Catch:
    打印日志("非JSON消息：" + 消息)
    Continue  // 跳过这条消息

什么时候用 HTTP，什么时候用 WebSocket

场景	推荐方案
查询一次数据，拿结果	HTTP GET
提交表单/创建数据	HTTP POST
轮询更新（每30秒查一次）	HTTP + 定时器
实时数据流（每秒多次更新）	WebSocket
服务器主动推送事件	WebSocket 或 SSE
双向实时通信	WebSocket