深入探究node搭建socket服务器

自从上篇中sokect实现了视频通话，但是是使用ws依赖库实现的服务端，所以最近再看ws源码，不看不知道，一看很惊讶。

接下来一点点记录一下，如何搭建一个简易的服务端socket，来实现上次的视频通讯。

搭建一个http服务

首先看一下ws依赖的调用

所以首选我们要创建一个服务器，然后监听端口号

这个不难，直接使用node自带的http依赖

const http = require('http');

class MyWebsocket extends EventEmitter {
  constructor(options) {
    super(options);

    const server = http.createServer();
    server.listen(options.port || 8080);
    
  }
}

module.exports = MyWebsocket;
1
2
3
4
5
6
7
8
9
10
11
12
13

这样就启动了一个端口号为8080的http服务了，然后这个端口可以自定义，所以初始化的时候，就传参进来就行。

然后我们继续发现，需要用on来监听事件，这要如何在node中实现呢？

on方法在这里遵循了Node.js EventEmitter模式，它允许我们绑定函数到特定的事件上，当该事件发生时，对应的函数会被执行。

什么意思呢？通熟易懂就是继承这个node自带的类EventEmitter

然后你要监听一个connection函数，在MyWebsocket中要怎么触发呢？

使用 emit 方法来触发你定义的事件，并传递任何你想要传递给监听器的数据。

const http = require('http');

class MyWebsocket extends EventEmitter {
  constructor(options) {
    super(options);

    const server = http.createServer();
    server.listen(options.port || 8080);
    this.emit('connection', 参数);
  }
}

module.exports = MyWebsocket;
1
2
3
4
5
6
7
8
9
10
11
12
13

如何监听客户端socket？

然后到了最重要的一步，我们最主要的功能就是监听socket，那怎么监听客户端来的socket连接？

看一下ws的websocket-server.js源码

我们刚刚不是建立了一个http服务吗？

监听 upgrade 事件

在 Node.js 中，HTTP 服务器可以监听 upgrade 事件来处理 WebSocket 连接或其他需要升级传输层协议的请求。upgrade 事件在客户端发起一个 HTTP 请求并要求升级到其他协议（如 WebSocket）时触发。

class MyWebsocket extends EventEmitter {
  constructor(options) {
    super();
    options = {
      ...options,
    }

    const server = http.createServer();
    server.listen(options.port || 8080);
    this.clients = new Set()
  

    server.on('upgrade', (req, socket) => {
      this.socket = socket; // 存储当前的socket，方便后端调用

      ...

    });
  }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

socket升级协议

然后需要有socket升级协议，为什么要有升级协议呢？

WebSocket 升级协议（WebSocket Upgrade Protocol）在 Node.js 中是必要的，因为它允许现有的 HTTP 或 HTTPS 服务器与客户端建立持久的、双向的通信连接，而这种连接在技术上被称为 WebSocket 连接。

那什么是socket升级协议呢？

1. 客户端请求：客户端发起一个 HTTP 请求，请求头部包含 Upgrade: websocket 和 Connection: Upgrade，以及可能的 Sec-WebSocket-Key 和其他 WebSocket 相关的头部信息。
2. 服务器响应：服务器接收到请求后，如果同意升级，会在响应中包含 Upgrade: websocket 和 Connection: Upgrade 头部，以及一个 Sec-WebSocket-Accept 头部，这个头部是服务器对客户端 Sec-WebSocket-Key 的回应。
3. 连接升级：一旦客户端和服务器都确认了升级，它们就会关闭 HTTP 连接，同时建立一个新的 WebSocket 连接。这个连接允许双方进行二进制或文本数据的双向通信。

其实就是根据客户端socket连接发过来的请求头，返回一个请求头给客户端来建立连接

看一下ws源码的处理

其实就说读取请求头中的sec-websocket-key字段，然后加上一个固定的字符串，经过 sha1 加密之后，转成 base64 的结果，就是digest

加密使用node中自带的crypto依赖

const crypto = require('crypto');

// 也就是用客户端传过来的 key，加上一个固定的字符串，经过 sha1 加密之后，转成 base64 的结果
function hashKey(key) {
  const sha1 = crypto.createHash('sha1');
  sha1.update(key + '258EAFA5-E914-47DA-95CA-C5AB0DC85B11');
  return sha1.digest('base64');
}
1
2
3
4
5
6
7
8

这个固定的字符串直接拿ws源码中的

然后就是升级协议的写入

const {
  EventEmitter
} = require('events');
const http = require('http');
const crypto = require('crypto');

const GUID = '258EAFA5-E914-47DA-95CA-C5AB0DC85B11'
// 也就是用客户端传过来的 key，加上一个固定的字符串，经过 sha1 加密之后，转成 base64 的结果
function hashKey(key) {
  const sha1 = crypto.createHash('sha1');
  sha1.update(key + GUID);
  return sha1.digest('base64');
}

class MyWebsocket extends EventEmitter {
  constructor(options) {
    super(options);

    const server = http.createServer();
    server.listen(options.port || 8080);

    server.on('upgrade', (req, socket) => {
      this.socket = socket;
      // socket.setKeepAlive(true);

      // websocket 升级协议
      const resHeaders = [
        'HTTP/1.1 101 Switching Protocols',
        'Upgrade: websocket',
        'Connection: Upgrade',
        'Sec-WebSocket-Accept: ' + hashKey(req.headers['sec-websocket-key']),
        '',
        ''
      ].join('\r\n');
      socket.write(resHeaders);

    });
  }

}

module.exports = MyWebsocket;
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

socket监听传输数据

接下来就说socket监听传输数据和socket关闭

socket.on('data', (data) => {
  console.log(data);
});
socket.on('close', (error) => {
  console.error('close', error)
});
1
2
3
4
5
6

然后我们一起看看效果吧

客户端发送的socket数据是

然后看请求头Sec-WebSocket-Accept也对应的上

处理socket传输数据

可以在node中拿到的数据是Buffer的二进制数据，首先需要处理的是WebSocket 协议中的数据帧。这里逻辑就有点复杂了。

协议中的数据帧结构是什么样子的？

数据帧的结构包括头部（Header）和负载（Payload）两部分。以下是数据帧的基本结构：

1. 控制位（Control Bits）：

• • FIN（1 bit）：表示这是消息的最后一个片段。如果为1，表示这是消息的结束；如果为0，表示还有后续片段。
  • RSV1、RSV2、RSV3（各1 bit）：保留位，用于未来的扩展，目前必须设置为0。
  • Opcode（4 bits）：操作码，定义了帧的类型。例如，0x1 表示文本帧，0x2 表示二进制帧，0x8 表示关闭连接，0x9 表示 Ping 帧，0xA 表示 Pong 帧等。
  • Mask（1 bit）：掩码位，指示负载数据是否被掩码。客户端发送给服务器的帧必须设置为1，表示数据被掩码；服务器发送给客户端的帧通常设置为0，表示数据未被掩码。

2. Payload Length（7、7+16、7+64 bits）：

• • 7位长度：如果值为0-125，表示负载数据的长度（以字节为单位）。
  • 7+16位长度：如果值为126，接下来的2个字节（16位）表示负载数据的长度。
  • 7+64位长度：如果值为127，接下来的8个字节（64位）表示负载数据的长度。

3. Masking-Key（0或4 bytes）：

• • 当掩码位（Mask）为1时，存在4字节的掩码密钥（Masking-Key）。这个密钥用于对负载数据进行掩码处理，以防止中间代理服务器缓存污染。

4. Payload Data（负载数据）：

• • 包含实际要传输的数据。对于文本帧，这是UTF-8编码的字符串；对于二进制帧，这是任意二进制数据。

从上面我们知道，需要的数据是负载数据，但是数据如果带有掩码，是需要解密的

1. 解析帧头：

• • 从 bufferData 的第一个字节（byte1）中读取操作码（opcode），这是一个4位的值，用于指示帧的类型（如文本、二进制等）。
  • 从第二个字节（byte2）中读取掩码位（MASK），这是一个1位的值，指示是否使用了掩码。

2. 计算有效载荷长度：

• • 如果 byte2 的最高位（第7位）是1，表示有效载荷长度为126，需要从 bufferData 的第3个字节和第4个字节（bufferData.readUInt16BE(2)）读取有效载荷的实际长度。
  • 如果 byte2 的最高位是0，但有效载荷长度为127，表示有效载荷长度为64位，需要从 bufferData 的第3个字节到第10个字节（bufferData.readBigUInt64BE(2)）读取有效载荷的实际长度。

3. 处理掩码：

• • 如果使用了掩码（MASK 为真），则从 bufferData 中提取掩码密钥（mask key），这是一个4字节的值。
  • 使用掩码密钥对有效载荷数据进行解密（通过 handleMask 函数），以获取实际的数据（realData）。

4. 处理有效载荷：

• • 最后，函数调用 handleRealData 方法，传入操作码和解密后的实际数据，进行进一步的处理。

function handleMask(maskBytes, data) {
  const payload = Buffer.alloc(data.length);
  for (let i = 0; i < data.length; i++) {
    payload[i] = maskBytes[i % 4] ^ data[i];
  }
  return payload;
}
const OPCODES = {
  CONTINUE: 0,
  TEXT: 1,
  BINARY: 2,
  CLOSE: 8,
  PING: 9,
  PONG: 10,
};


class MyWebsocket extends EventEmitter {
  constructor(options) {
    ...
  }
// 处理 WebSocket 协议中的数据帧
  processData(bufferData) {
    const byte1 = bufferData.readUInt8(0); // 第一个字节（byte1）中读取操作码（opcode），这是一个4位的值，用于指示帧的类型（如文本、二进制等）。
    let opcode = byte1 & 0x0f; 
    
    const byte2 = bufferData.readUInt8(1); // 从第二个字节（byte2）中读取掩码位（MASK），这是一个1位的值，指示是否使用了掩码。
    const str2 = byte2.toString(2);
    const MASK = str2[0];
    console.log(opcode, 'opcode')
    console.log(MASK, 'mask')

    let curByteIndex = 2;
    
    let payloadLength = parseInt(str2.substring(1), 2);
    if (payloadLength === 126) {
      payloadLength = bufferData.readUInt16BE(2);
      curByteIndex += 2;
    } else if (payloadLength === 127) {
      payloadLength = bufferData.readBigUInt64BE(2);
      curByteIndex += 8;
    }
    console.log(payloadLength, 'payloadLength')
    let realData = null;
    
    if (MASK) {
      const maskKey = bufferData.slice(curByteIndex, curByteIndex + 4); // 掩码密钥
      curByteIndex += 4;
      const payloadData = bufferData.slice(curByteIndex, curByteIndex + payloadLength);
      realData = handleMask(maskKey, payloadData); // 使用掩码密钥对有效载荷数据进行解密，以获取实际的数据（realData）。
    } 
    console.log(realData, 'realData')
    this.handleRealData(opcode, realData); // 处理有效载荷
  }
handleRealData(opcode, realDataBuffer) {
    switch (opcode) {
      case OPCODES.TEXT: // 文本
        this.emit('data', realDataBuffer);
        break;
      case OPCODES.BINARY: // 二进制
        this.emit('data', realDataBuffer);
        break;
      default:
        this.emit('close');
        break;
    }
  }

  handleRealData(opcode, realDataBuffer) {
    switch (opcode) {
      case OPCODES.TEXT: // 文本
        this.emit('data', realDataBuffer);
        break;
      case OPCODES.BINARY: // 二进制
        this.emit('data', realDataBuffer);
        break;
      default:
        this.emit('close');
        break;
    }
  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83

然后调用main.js

const MyWebSocket = require('./ws.js');

const ws = new MyWebSocket({ port: 8000 });
// websocket需要一个服务器，如果两个客户端需要通讯，需要用服务器转发\


ws.on('data', (data) => {
  console.log('receive data:' + data); // 接受消息
});
1
2
3
4
5
6
7
8
9

可以看到，存在掩码，解密之前数据是bufferData，解密之后的数据是realData

这样就成功拿到了客户端传过来的数据了，可以看到客户端传过来的是文本，使用了掩码，效载荷长度为9位，这里的9其实就说字符串{“A”:111}的长度。

服务端发消息给客户端

服务端能接收到消息了，然后就是将消息再给客户端了，所以需要定义一个函数，来发送数据

class MyWebsocket extends EventEmitter {
  constructor(options) {
    ...
  }
  ...
  send(data) {
    let opcode;
    let buffer;
    if (Buffer.isBuffer(data)) {
      opcode = OPCODES.BINARY;
      buffer = data;
    } else if (typeof data === 'string') {
      opcode = OPCODES.TEXT;
      buffer = Buffer.from(data, 'utf8');
    } else {
      console.error('暂不支持发送的数据类型')
    }
    this.doSend(opcode, buffer);
  }

  doSend(opcode, bufferDatafer) {
    this.socket.write(encodeMessage(opcode, bufferDatafer));
  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

由于我们上面获取传输数据的时候，知道socket数据需要支持WebSocket 协议中的数据帧的帧结构

因为根据 WebSocket 协议，只有客户端发送给服务器的帧需要掩码。服务器发送给客户端的帧通常不需要掩码。

function encodeMessage(opcode, payload) {
  let bufferData = Buffer.alloc(payload.length + 2 + 0);

  let byte1 = parseInt('10000000', 2) | opcode; // parseInt(130, 2)=1 ; 设置 FIN 为 1
  let byte2 = payload.length;

  bufferData.writeUInt8(byte1, 0); // 
  bufferData.writeUInt8(byte2, 1); // 负载的长度

  payload.copy(bufferData, 2);

  return bufferData;
}
1
2
3
4
5
6
7
8
9
10
11
12
13

1. 创建缓冲区：

• • 使用 Buffer.alloc 方法创建一个足够大的 Buffer 对象，以容纳操作码、有效载荷长度和实际的有效载荷数据。这里假设 payload.length < 126，所以有效载荷长度只需要1个字节来表示。

2. 设置操作码：

• • byte1 是第一个字节，它包含了操作码和 FIN（Finish）标志。这里假设 FIN 标志为 1（即消息结束），操作码通过 opcode 参数传入。操作码的值决定了消息的类型，例如文本（0x1）或二进制（0x2）。

3. 设置有效载荷长度：

• • byte2 是第二个字节，它包含了有效载荷的长度。由于有效载荷长度小于126，所以只需要1个字节来表示。

4. 写入操作码和有效载荷长度：

• • 使用 writeUInt8 方法将 byte1 和 byte2 分别写入 bufferData 的第0位和第1位。

5. 复制有效载荷数据：

• • 使用 copy 方法将 payload 数据复制到 bufferData 的第2位及之后的位置。

const MyWebSocket = require('./ws.js');

const ws = new MyWebSocket({ port: 8000 });
// websocket需要一个服务器，如果两个客户端需要通讯，需要用服务器转发\


ws.on('data', (data) => {
  console.log('receive data:' + data); // 接受消息
  ws.send(data); // 自己给自己发送消息
});
1
2
3
4
5
6
7
8
9
10

客户端接收到的数据

// 创建WebSocket连接
const socketA = new WebSocket('ws://localhost:8000');

const handleBlobToText = (blob) => {
  let reader = new FileReader()
  reader.readAsText(blob, 'utf-8') // 接收到的是blob数据，先转成文本
  reader.onload = async function () {
    console.log(reader.result)
  }
}
// A接收B的消息
socketA.onmessage = function (event) {
  console.log('A received:', event.data);
  handleBlobToText(event.data)
};
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

socket传输大量数据

然后直接将视频的数据，传输给服务端，然后服务端就挂了