RStudio中WebSocket控制台输出与onMessage函数结果保存方法

实现方案 #

不需要改动原有鉴权、心跳处理的核心逻辑，仅在原有回调基础上补全存储逻辑即可，核心要解决R回调的作用域赋值失效、断连丢数、非目标消息混存三个实操问题。

第一步：初始化全局存储容器 #

WebSocket包的回调函数运行在独立的临时执行环境中，使用普通<-做的局部赋值，会在回调执行完成后被系统回收，无法跨消息积累历史数据。因此需要提前在全局环境定义两个持久化存储对象：

# 存储全量连接日志，覆盖所有连接阶段的内容
all_connection_logs <- data.frame(
  capture_time = as.POSIXct(character()),
  content = character(),
  stringsAsFactors = FALSE
)

# 单独存储含Price1/2/3的行情Tick数据
price_tick_data <- data.frame(
  capture_time = as.POSIXct(character()),
  instrument = character(),
  seq_number = integer(),
  price1 = numeric(),
  price2 = numeric(),
  price3 = numeric(),
  stringsAsFactors = FALSE
)

第二步：全量留存连接过程的所有内容 #

全量日志需要覆盖两类易遗漏的内容：

• 连接建立、登录请求发送的阶段日志：这类内容不是onMessage事件触发的，需要单独绑定onOpen回调捕获
• 所有onMessage收到的原始报文：需要在回调最开始就做留存，防止后续解析出错丢失原始数据

对应代码实现：

# 绑定连接建立事件，补全初始阶段日志
ws$onOpen(function(event) {
  current_time <- Sys.time()
  # 追加连接建立、发送登录请求的日志
  log_block <- data.frame(
    capture_time = current_time,
    content = c("Connection opened", "send login request"),
    stringsAsFactors = FALSE
  )
  # 用<<-赋值到全局存储对象，避免局部赋值失效
  all_connection_logs <<- rbind(all_connection_logs, log_block)
  # 实时写入本地文件，防止R进程退出导致内存数据丢失
  write.csv(all_connection_logs, "ws_full_connection_log.csv", row.names = FALSE)
})

# 重写onMessage回调，在原有逻辑基础上加存储能力
ws$onMessage(function(event) {
  current_time <- Sys.time()
  raw_received_msg <- event$data

  # 1. 第一时间留存原始消息到全量日志
  log_entry <- data.frame(
    capture_time = current_time,
    content = raw_received_msg,
    stringsAsFactors = FALSE
  )
  all_connection_logs <<- rbind(all_connection_logs, log_entry)
  write.csv(all_connection_logs, "ws_full_connection_log.csv", row.names = FALSE)

  # 保留原有控制台输出逻辑
  cat("Got Message", raw_received_msg, "\n")

  # 保留原有消息解析、心跳处理逻辑
  jsonresp <- fromJSON(raw_received_msg, simplifyDataFrame = FALSE)
  for (single_msg in jsonresp) {
    process_message(single_msg)

    # 2. 定向提取含Price1-3的行情数据
    # 仅处理带Fields节点的服务器返回消息，自动跳过客户端发出的订阅请求类消息
    if (!is.null(single_msg$Fields)) {
      field_content <- single_msg$Fields
      # 校验是否存在三个目标价格字段
      if (all(c("Price1", "Price2", "Price3") %in% names(field_content))) {
        tick_entry <- data.frame(
          capture_time = current_time,
          instrument = single_msg$Key$Name,
          seq_number = single_msg$SeqNumber,
          price1 = field_content$Price1,
          price2 = field_content$Price2,
          price3 = field_content$Price3,
          stringsAsFactors = FALSE
        )
        price_tick_data <<- rbind(price_tick_data, tick_entry)
        # 行情数据单独持久化
        write.csv(price_tick_data, "ws_price_tick.csv", row.names = FALSE)
      }
    }
  }
})

实操注意事项 #

• 之前尝试用test1 <- event$data存内容的写法存在根本缺陷：这种局部赋值会被新的消息不断覆盖，永远存不下全量历史数据，必须使用<<-将内容赋值到全局的存储对象上才能积累记录。
• 每次收到消息就实时写盘，是因为WebSocket属于长连接，一旦网络断开、R进程异常退出，仅存在内存里的未保存数据会全部丢失，落盘操作可以保证已经收到的内容不丢失。
• 代码里的字段判断逻辑会自动跳过发出的订阅请求消息（比如{"ID":2,"Key":{"Name":"XXX"},"View":["Price1","Price2","Price3"]}），这类消息没有Fields节点，不会被误存到行情数据文件里。
• 如果后续行情推送频率很高，可以把写盘逻辑调整为攒够10-20条记录再批量写入，降低IO开销提升性能。

内容的提问来源于stack exchange，提问作者Erko Tru