CR1000X与NodeJS服务器TCP连接异常断开的CRBasic修复方案咨询

问题描述 #

我正在编写CRBasic程序，让CR1000X数据采集器向NodeJS TCP服务器发送数据。现有NodeJS服务器代码可接收数据，但偶尔会出现ECONNRESET错误导致连接断开。当前CRBasic代码仅初始化一次Socket，需要在CRBasic中捕获Socket异常，实现自动重连并补发数据。

原NodeJS服务器代码 #

const net = require('net');

// 创建TCP服务器
const server = net.createServer((socket) => {
  console.log('Client connected:', socket.remoteAddress, socket.remotePort);

  socket.on('data', (data) => {
    console.log('Hex Data:', data.toString('utf-8'));
  });

  socket.on('end', () => {
    console.log('Client disconnected');
  });
  socket.on('error', error => {
    console.log('Client disconnected', error);
  });
});

const PORT = 6785;
server.listen(PORT, () => {
  console.log(`Server is listening on port ${PORT}`);
});

断开错误日志 #

Client disconnected Error: read ECONNRESET
    at TCP.onStreamRead (node:internal/stream_base_commons:216:20) {
  errno: -4077,
  code: 'ECONNRESET',
  syscall: 'read'
}

原CRBasic代码 #

BeginProg
  Socket = TCPOpen(IP_Addr,PORT,3)
  CPISpeed(250)            '250 kbps, how fast the datalogger will operate the CPI bus.
  'This value may need to be adjusted based on the number of
  'VWIRE 305 modules and the total cable length.  Refer to the
  'VWIRE 305 quick start guide for more information.

  Scan(5, msec, 500, 0)    '200 Hz / 5 msec scan rate
    CDM_VW300Dynamic(CPIAddress, Freq(), Diag())   'Get dynamic readings
    CallTable DynamicFreq

    If TimeIntoInterval(0,1,Sec) Then ' Process static data only once per second
      CDM_VW300Static(CPIAddress, StaticFreq(), Therm(), DynStdDev()) 'Get static data
      CallTable OneSecData

      OutString = FormatFloat(Freq(1), "%f")
      OutString = OutString & "," & FormatFloat (Freq(2), "%f") 

      For i = 1 To 2
        OutString = OutString & "," & FormatFloat (Therm(i), "%f") 
      Next

      If Socket > 0 Then
        SerialOut(Socket, OutString, 0, 1, 0)
      End If
    EndIf
  NextScan

EndProg

解决方案 #

一、CRBasic代码核心修改 #

要实现异常捕获、自动重连和数据补发，需加入连接状态检测、数据缓存和重连重试逻辑，具体修改如下：

关键改进点

• 每次发送前检查Socket连接有效性，无效则触发重连
• 用数组缓存未成功发送的数据，重连成功后优先补发
• 增加重连重试机制，避免单次失败直接放弃
• 捕获SerialOut返回值判断发送是否成功

修改后的CRBasic代码

' 定义全局变量：Socket句柄、数据缓存数组、缓存索引
Dim Socket As Long
Dim DataBuffer(100) As String ' 最多缓存100条数据，可根据需求调整
Dim BufferIndex As Long

BeginProg
  BufferIndex = 0 ' 初始化缓存索引
  Socket = -1 ' 初始Socket状态设为无效
  
  CPISpeed(250)            '250 kbps, how fast the datalogger will operate the CPI bus.
  'This value may need to be adjusted based on the number of
  'VWIRE 305 modules and the total cable length.  Refer to the
  'VWIRE 305 quick start guide for more information.

  Scan(5, msec, 500, 0)    '200 Hz / 5 msec scan rate
    CDM_VW300Dynamic(CPIAddress, Freq(), Diag())   'Get dynamic readings
    CallTable DynamicFreq

    If TimeIntoInterval(0,1,Sec) Then ' Process static data only once per second
      CDM_VW300Static(CPIAddress, StaticFreq(), Therm(), DynStdDev()) 'Get static data
      CallTable OneSecData

      ' 组装要发送的数据
      OutString = FormatFloat(Freq(1), "%f")
      OutString = OutString & "," & FormatFloat (Freq(2), "%f") 

      For i = 1 To 2
        OutString = OutString & "," & FormatFloat (Therm(i), "%f") 
      Next
      OutString = OutString & "\n" ' 添加换行符作为数据分隔符，配合服务器解析

      ' ====================== 连接状态检查与重连逻辑 ======================
      If Socket <= 0 Then
        ' 尝试重新建立TCP连接，最多重试3次
        For RetryCnt = 1 To 3
          Socket = TCPOpen(IP_Addr, PORT, 3) ' 第三个参数是超时时间（秒）
          If Socket > 0 Then
            ' 重连成功，补发缓存的数据
            For BufferPos = 1 To BufferIndex
              SerialOut(Socket, DataBuffer(BufferPos), 0, 1, 0)
              Delay(100) ' 补发时加短暂延迟，避免服务器压力过大
            Next
            BufferIndex = 0 ' 补发完成后清空缓存
            Exit For ' 跳出重试循环
          End If
          Delay(1000) ' 重试间隔1秒
        Next
      End If

      ' ====================== 数据发送与缓存逻辑 ======================
      If Socket > 0 Then
        ' 尝试发送数据，捕获返回值判断是否成功
        SendResult = SerialOut(Socket, OutString, 0, 1, 0)
        If SendResult <= 0 Then ' 发送失败（连接断开或其他错误）
          ' 将数据存入缓存（如果缓存未满）
          If BufferIndex < UBound(DataBuffer) Then
            BufferIndex = BufferIndex + 1
            DataBuffer(BufferIndex) = OutString
          End If
          Socket = -1 ' 标记Socket为无效，下次扫描触发重连
        End If
      Else
        ' Socket无效，直接存入缓存
        If BufferIndex < UBound(DataBuffer) Then
          BufferIndex = BufferIndex + 1
          DataBuffer(BufferIndex) = OutString
        End If
      End If
    EndIf
  NextScan

EndProg

二、NodeJS服务器优化（减少ECONNRESET触发） #

ECONNRESET通常由半开连接、超时或异常断开导致，可通过以下优化减少错误：

1. 增加心跳机制，检测无效连接

定期发送心跳包，若客户端无响应则主动关闭连接：

const net = require('net');

const server = net.createServer((socket) => {
  console.log('Client connected:', socket.remoteAddress, socket.remotePort);
  
  // 心跳定时器：每30秒发送一次心跳
  const heartbeatInterval = setInterval(() => {
    if (socket.writable) {
      socket.write('PING\n');
    } else {
      clearInterval(heartbeatInterval);
      socket.destroy();
    }
  }, 30000);

  socket.on('data', (data) => {
    const dataStr = data.toString('utf-8');
    // 处理心跳响应
    if (dataStr.trim() === 'PONG') {
      return;
    }
    // 按换行符拆分数据，处理粘包
    const messages = dataStr.split('\n').filter(msg => msg.trim() !== '');
    messages.forEach(msg => {
      console.log('Received Data:', msg);
    });
  });

  socket.on('end', () => {
    clearInterval(heartbeatInterval);
    console.log('Client disconnected');
  });
  
  socket.on('error', error => {
    clearInterval(heartbeatInterval);
    console.log('Client error:', error);
  });
});

const PORT = 6785;
server.listen(PORT, () => {
  console.log(`Server is listening on port ${PORT}`);
});

（注：需在CRBasic中添加心跳响应逻辑，收到PING\n时返回PONG\n）

2. 处理TCP粘包问题

TCP是流协议，可能出现多条数据粘在一起的情况，通过约定换行符\n拆分数据，确保每条数据独立解析。

关键说明 #

• 缓存数组大小可根据数据发送频率和重连耗时调整，避免溢出
• 重连间隔和重试次数可根据网络环境优化，过短可能导致资源浪费
• SerialOut返回值大于0表示发送成功，<=0表示失败（包括连接断开）

内容的提问来源于stack exchange，提问作者Dung Do