Node.js UDP 可靠性实战：丢包乱序分析与可靠传输协议设计

前言

UDP 协议在常规开发中较少涉及，与 TCP 不同，它被定义为“不可靠”传输。本文将通过实际案例来验证其不可靠的具体表现。

不可靠的 UDP

在实验开始前，先介绍所需的工具（以 Mac 环境为例，其他环境请自行查找对应工具）：

Network Link Conditioner：用于模拟网络丢包场景，可从苹果开发者官网下载。
Wireshark：用于网络抓包与分析的工具。
云主机：由于发现 Network Link Conditioner 对本地回环地址无效，因此需要借助云主机进行测试，若有其他更优方案欢迎指正。

接下来准备两段代码，分别作为 UDP Server 和 UDP Client。Client 将向 Server 发送 26 个大写英文字母，Server 负责将这些字母保存到文件中：

// udp-server.js
const udp = require('dgram')
const server = udp.createSocket('udp4')
const fs = require('fs')
server.on('listening', function () {
  var address = server.address()
  var port = address.port
  console.log('Server is listening at port ' + port)
})
server.on('message', function (msg, info) {
  console.log(
    `Data received from ${info.address}:${info.port}: ${msg.toString()}`
  )
  fs.appendFileSync('./out', msg.toString())
})
server.on('error', function (error) {
  console.log('Error: ' + error)
  server.close()
})
server.bind(7788)
// udp-client.js
const udp = require('dgram')
const client = udp.createSocket('udp4')
for (let i = 0; i < 26; i++) {
  const char = String.fromCharCode(0x41 + i)
  client.send(Buffer.from(char), 7788, '********', function (error) {
    if (error) {
      console.log(error)
    }
  })
}

随后按照以下步骤进行实验：

通过 Network Link Conditioner 将丢包率设置为 50%：

配置 Wireshark 的抓包参数：

在云主机上启动 Server，在本地启动 Client。

接着，我们观察实验结果：

首先，服务端接收到的字母数量明显减少，仅收到 14 个：

服务端接收到的字母顺序出现错乱，例如 U 出现在 T 之前：

为了进行对比，我们可以尝试使用 TCP，代码如下，此处省略结果展示：

// tcp-server.js
const net = require('net')
const server = net.createServer()
const fs = require('fs')
server.on('connection', function (conn) {
  conn.on('data', (msg) => {
    console.log(
      `Data received from ${conn.address().address}:${
        conn.address().port
      }: ${msg.toString()}`
    )
    fs.appendFileSync('./out', msg.toString())
  })
})
server.listen(8899, () => {
  console.log('server listening to %j', server.address().port)
})
// tcp-client.js
var net = require('net')
var client = new net.Socket()
client.connect(8899, '********', function () {
  for (let i = 0; i < 26; i++) {
    const char = String.fromCharCode(0x41 + i)
    client.write(char)
  }
})

接下来，我们将尝试基于 UDP 实现一个可靠的传输协议，主要解决上述的丢包和乱序问题。

基于 UDP 的简单可靠传输协议

首先，需要设计协议格式。为了简化实现，我们在原有 UDP 数据部分新增 4 个字节的 SEQ（序列号）和 4 个字节的 ACK（确认号）：

+-------------------------------+
|      64 个字节的 UDP 首部       |
+-------------------------------+
|  SEQ(4 个字节) |  ACK(4 个字节) |
+-------------------------------+
|             Data              |
+-------------------------------+

其中，SEQ 表示当前数据包的序号，ACK 表示回复确认的序号。

接下来，我们看看如何解决前文提到的两个问题。

乱序问题

接收方需要维护一个名为 expectedSeq 的变量，用于表示期待接收到的包序号。为简化逻辑，制定如下规则：如果当前接收到的包序号等于 expectedSeq，则将包交给应用层处理，并向发送方发送 ACK；否则直接丢弃。当然，更优的做法是维护一个接收窗口，这样可以批量提交数据给应用层，并缓存大于 expectedSeq 的包。

假设发送方发送了 1、2、3 两个包（注：原文表述可能有误，通常指多个包），但到达接收方的顺序是 3、2、1。按照上述规则，接收方会丢弃 3 和 2，仅接收 1。虽然顺序不再混乱，但数据包丢失了。

因此，还需要解决丢包问题。

丢包问题

发送方维护一个发送窗口，用于存储已发送但尚未被确认的包：

+---+---+---+---+
| 1 | 2 | 3 | 4 |
+---+---+---+---+

发送方每发送一个包时，需将该包放入发送窗口，并设置定时器以进行重发。当发送方接收到来自接收方的 ACK 时，需取消对应包的定时器，并将发送窗口中序号小于 ACK 的包全部删除。

+---+---+---+---+
| 1 | 2 | 3 | 4 |
+---+---+---+---+
// ACK = 4，删除 1 2 3，并取消掉他们的定时器
+---+
| 4 |
+---+

完整代码及使用 Demo 见文末，现在可以正常按顺序输出 26 个字母了，但距离真正的“可靠”协议仍有差距。例如，第一次输出完 26 个字母后，再次启动客户端时发现没有任何输出。原因在于此时接收端的 expectedSeq 已经增加到 20 以上，但新启动的 Client 发送的 SEQ 仍从 1 开始，导致接收端一直丢弃接收到的包，而发送端不断重试。

要解决此问题，可参考 TCP 在传输两端建立“连接”的概念。在开始发送前，通过“三次握手”建立连接，确定起始 SEQ 并初始化窗口；结束前通过“四次挥手”断开连接，清理窗口定时器等工作。这部分内容留待后续讨论。

代码

// packet.js
class Packet {
  constructor({seq, ack, data = ''}) {
    this.seq = seq // 序列号
    this.ack = ack // 确认号
    this.data = data // 数据
  }
  // 将 Packet 转换成 Buffer，以便通过网络传输
  toBuffer() {
    const seqBuffer = Buffer.alloc(4)
    seqBuffer.writeUInt32BE(this.seq)
    const ackBuffer = Buffer.alloc(4)
    ackBuffer.writeUInt32BE(this.ack)
    const dataBuffer = Buffer.from(this.data)
    return Buffer.concat([seqBuffer, ackBuffer, dataBuffer])
  }
  // 从 Buffer 中解析出 Packet
  static fromBuffer(buffer) {
    const seq = buffer.readUInt32BE()
    const ack = buffer.readUInt32BE(4)
    const data = buffer.slice(8)
    return new Packet({seq, ack, data})
  }
}
module.exports = Packet
// reliableUDP.js
const dgram = require('dgram')
const Packet = require('./packet')
class ReliableUDP {
  constructor() {
    this.socket = dgram.createSocket('udp4')
    this.socket.on('message', this.handleMessage.bind(this))
    this.sendWindow = [] // 发送窗口，用于存放待确认的数据包
    this.receiveWindow = [] // 接收窗口，用于存放已接收的数据包
    this.expectedSeq = 1 // 期望接收的数据包序列号
    this.nextSeq = 1 // 下一个要发送的数据包序列号
    this.timeout = 100 // 超时时间，单位为毫秒
    this.timeoutIds = {} // 用于存放定时器 ID
  }
  listen(port, address, fn) {
    this.socket.bind(port, address, fn)
  }
  // 发送数据包
  sendPacket(packet, address, port) {
    const buffer = packet.toBuffer()
    this.socket.send(buffer, port, address, (err) => {
      if (err) {
        console.error(err)
      }
    })
    if (packet.ack) return
    if (!this.sendWindow.includes((p) => p.seq === packet.seq))
      this.sendWindow.push(packet)
    // 设置超时定时器
    const timeoutId = setTimeout(() => {
      this.handleTimeout(packet.seq, address, port)
    }, this.timeout)
    this.timeoutIds[packet.seq] = timeoutId
  }
  // 处理接收到的数据包
  handleMessage(msg, rinfo) {
    const {address, port} = rinfo
    const packet = Packet.fromBuffer(msg)
    // 收到的是应答的包
    if (packet.ack) {
      const ackNum = packet.ack - 1
      // 处理发送窗口中已经确认的数据包
      while (this.sendWindow.length > 0 && this.sendWindow[0].seq <= ackNum) {
        this.sendWindow.shift()
      }
      // 清除超时定时器
      if (this.timeoutIds[ackNum]) {
        clearTimeout(this.timeoutIds[ackNum])
        delete this.timeoutIds[ackNum]
      }
    } else {
      // 如果是重复的数据包，则忽略
      if (packet.seq < this.expectedSeq) {
        return
      }
      // 如果是期望接收的数据包
      if (packet.seq === this.expectedSeq) {
        this.receiveWindow.push(packet)
        this.expectedSeq++
        // 处理接收窗口中已经确认的数据包
        while (
          this.receiveWindow.length > 0 &&
          this.receiveWindow[0].seq <= this.expectedSeq
        ) {
          const packet = this.receiveWindow.shift()
          this.onPacketReceived(packet.data)
        }
        const ackPacket = new Packet({
          seq: this.nextSeq++,
          ack: this.expectedSeq,
        })
        this.sendPacket(ackPacket, address, port)
      } else {
        // 如果是未来的数据包，暂不做处理，更好的做法是缓存起来
      }
    }
  }
  // 应用层调用该方法发送数据
  send(data, address, port) {
    const packet = new Packet({
      seq: this.nextSeq,
      ack: null,
      data,
    })
    this.sendPacket(packet, address, port)
    this.nextSeq++
  }
  // 应用层调用该方法注册回调函数，接收数据
  onReceive(callback) {
    this.onPacketReceived = callback
  }
  // 处理超时
  handleTimeout(seq, address, port) {
    // 重传超时的数据包
    const packet = this.sendWindow.find((p) => p.seq === seq)
    if (packet) {
      this.sendPacket(packet, address, port)
    }
  }
}
module.exports = ReliableUDP
// server.js
const ReliableUDP = require('./reliableUDP')
const server = new ReliableUDP()
server.listen(7788, 'localhost')
server.onReceive((data) => {
  console.log(data.toString())
})
// client.js
const ReliableUDP = require('./reliableUDP')
const client = new ReliableUDP()
for (let i = 0; i < 26; i++) {
  const char = String.fromCharCode(0x41 + i)
  client.send(char, 'localhost', 7788)
}

以上就是 Node.js 高级编程之 UDP 可靠性分析的详细内容，更多关于 Node.js 高级编程 UDP 的资料请关注本站其它相关文章！

Node.js 基于 UDP 的可靠传输协议实现与实验分析

前言

不可靠的 UDP

基于 UDP 的简单可靠传输协议

乱序问题

丢包问题

代码

实用工具推荐