深入探究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;

这样就启动了一个端口号为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;

如何监听客户端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,方便后端调用...});}}

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');
}

这个固定的字符串直接拿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;

socket监听传输数据

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

socket.on('data', (data) => {console.log(data);
});
socket.on('close', (error) => {console.error('close', error)
});

然后我们一起看看效果吧

客户端发送的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,表示数据未被掩码。
  1. Payload Length(7、7+16、7+64 bits):
    • 7位长度:如果值为0-125,表示负载数据的长度(以字节为单位)。
    • 7+16位长度:如果值为126,接下来的2个字节(16位)表示负载数据的长度。
    • 7+64位长度:如果值为127,接下来的8个字节(64位)表示负载数据的长度。
  1. Masking-Key(0或4 bytes):
    • 当掩码位(Mask)为1时,存在4字节的掩码密钥(Masking-Key)。这个密钥用于对负载数据进行掩码处理,以防止中间代理服务器缓存污染。
  1. Payload Data(负载数据):
    • 包含实际要传输的数据。对于文本帧,这是UTF-8编码的字符串;对于二进制帧,这是任意二进制数据。

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

  1. 解析帧头:
    • 从 bufferData 的第一个字节(byte1)中读取操作码(opcode),这是一个4位的值,用于指示帧的类型(如文本、二进制等)。
    • 从第二个字节(byte2)中读取掩码位(MASK),这是一个1位的值,指示是否使用了掩码。
  1. 计算有效载荷长度:
    • 如果 byte2 的最高位(第7位)是1,表示有效载荷长度为126,需要从 bufferData 的第3个字节和第4个字节(bufferData.readUInt16BE(2))读取有效载荷的实际长度。
    • 如果 byte2 的最高位是0,但有效载荷长度为127,表示有效载荷长度为64位,需要从 bufferData 的第3个字节到第10个字节(bufferData.readBigUInt64BE(2))读取有效载荷的实际长度。
  1. 处理掩码:
    • 如果使用了掩码(MASK 为真),则从 bufferData 中提取掩码密钥(mask key),这是一个4字节的值。
    • 使用掩码密钥对有效载荷数据进行解密(通过 handleMask 函数),以获取实际的数据(realData)。
  1. 处理有效载荷:
    • 最后,函数调用 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;}}
}

然后调用main.js

const MyWebSocket = require('./ws.js');const ws = new MyWebSocket({ port: 8000 });
// websocket需要一个服务器,如果两个客户端需要通讯,需要用服务器转发\ws.on('data', (data) => {console.log('receive data:' + data); // 接受消息
});

可以看到,存在掩码,解密之前数据是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));}
}

由于我们上面获取传输数据的时候,知道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 为 1let byte2 = payload.length;bufferData.writeUInt8(byte1, 0); // bufferData.writeUInt8(byte2, 1); // 负载的长度payload.copy(bufferData, 2);return bufferData;
}
  1. 创建缓冲区:
    • 使用 Buffer.alloc 方法创建一个足够大的 Buffer 对象,以容纳操作码、有效载荷长度和实际的有效载荷数据。这里假设 payload.length < 126,所以有效载荷长度只需要1个字节来表示。
  1. 设置操作码:
    • byte1 是第一个字节,它包含了操作码和 FIN(Finish)标志。这里假设 FIN 标志为 1(即消息结束),操作码通过 opcode 参数传入。操作码的值决定了消息的类型,例如文本(0x1)或二进制(0x2)。
  1. 设置有效载荷长度:
    • byte2 是第二个字节,它包含了有效载荷的长度。由于有效载荷长度小于126,所以只需要1个字节来表示。
  1. 写入操作码和有效载荷长度:
    • 使用 writeUInt8 方法将 byte1 和 byte2 分别写入 bufferData 的第0位和第1位。
  1. 复制有效载荷数据:
    • 使用 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); // 自己给自己发送消息
});

客户端接收到的数据

// 创建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)
};

socket传输大量数据

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

可以看到,node端是收到了客户端的数据

报错的原因是超出了范围,原因就是我们发送消息给客户端的处理这里出现了问题,也就是encodeMessage函数。

我们往前看看,到处理socket传输数据中,再仔细看看数据帧结构,这里有关于负载长度的问题

我们知道,我们需要处理的负载,就是我们需要传输的数据,然后数据量太大,是要区分来处理数据的。

很明显,上面的encodeMessage只适用于处理0-125的负载长度,而发送视频的数据,我们可以看看长度为多少

找到问题了,需要解决一下,接下来改写一下encodeMessage函数

你需要考虑 WebSocket 的最大帧大小限制。WebSocket 协议定义了三种帧类型来表示数据的长度:

  1. 单字节帧:用于长度小于 126 的数据。
  2. 双字节帧:用于长度在 126 到 65535 之间的数据。
  3. 八字节帧:用于长度大于 65535 的数据。
function encodeMessagePerf(options, data) {let offset = 2;let dataLength = data.length;let payloadLength = dataLength;// WebSocket 的最大帧大小限制// 1. 单字节帧:用于长度小于 126 的数据。// 2. 双字节帧:用于长度在 126 到 65535 之间的数据。// 3. 八字节帧:用于长度大于 65535 的数据。if (dataLength >= 65536) {offset += 8;payloadLength = 127;} else if (dataLength > 125) {offset += 2;payloadLength = 126;}const target = Buffer.allocUnsafe(offset);// 操作码 0x1表示文本帧;0x2表示二进制帧;0x8表示关闭连接;0x9表示ping帧;0xA表示pong帧。target[0] = options | 0x80; // 设置FINtarget[1] = payloadLength; // 负载长度if (payloadLength === 126) {target.writeUInt16BE(dataLength, 2);} else if (payloadLength === 127) {target[2] = target[3] = 0;target.writeUIntBE(dataLength, 4, 6);}// 根据 WebSocket 协议,只有客户端发送给服务器的帧需要掩码。服务器发送给客户端的帧通常不需要掩码return [target, data];
}

writeUInt16BE、writeUIntBE 和 writeUInt8 是 Node.js 中 Buffer 类的三个方法,它们的主要区别在于它们写入的值的大小和字节序。

  1. writeUInt16BE:
    • 这个方法用于将一个无符号的16位整数(即0到65535之间的整数)以大端字节序(Big Endian)写入到 Buffer 对象中。大端字节序意味着高位字节在前,低位字节在后。
    • 例如,如果你要写入的值是 0xABCD,使用 writeUInt16BE 方法后,Buffer 中的数据将是 0xAB 后跟 0xCD。
    • 使用这个方法时,你需要指定一个 offset 参数,表示从 Buffer 的哪个位置开始写入。如果 offset 超出了 Buffer 的长度,或者提供的值不是有效的无符号16位整数,行为是未定义的。
  1. writeUInt8:
    • 这个方法用于将一个无符号的8位整数(即0到255之间的整数)写入到 Buffer 对象中。这个方法不涉及字节序,因为它只处理一个字节。
    • 使用 writeUInt8 方法时,你同样需要指定一个 offset 参数。如果 offset 超出了 Buffer 的长度,或者提供的值不是有效的无符号8位整数,行为同样是未定义的。
  1. writeUIntBE:

用于将无符号整数以大端序(Big Endian)格式写入到 Buffer 对象中,它可以处理的整数大小可以达到 48 位(6 个字节)。这个方法允许你指定要写入的字节长度(byteLength),这可以是 1、2、3、4、5 或 6 字节。如果 byteLength 大于 2,它将写入一个大于 16 位的整数。

总结来说,writeUInt16BE 用于写入16位的整数,并且遵循大端字节序,而 writeUInt8 用于写入8位的整数,不涉及字节序。两者都需要一个 offset 参数来指定写入位置。在实际应用中,选择哪个方法取决于你需要写入的数据类型和字节序要求。如果你需要写入更大或更小的整数,或者需要处理可变长度的整数,那么 writeUIntBE 是更合适的选择。

Buffer.allocUnsafe 和 Buffer.alloc 是 Node.js 中用于创建 Buffer 实例的两种方法,它们的主要区别在于内存的初始化方式和安全性。

  1. Buffer.allocUnsafe(size):
    • Buffer.allocUnsafe 创建一个指定大小的 Buffer 实例,但它不会初始化分配的内存。这意味着,新创建的 Buffer 实例可能包含之前内存中的数据,这些数据可能是敏感的。因此,这个方法被称为“不安全”(unsafe)。
    • 使用 Buffer.allocUnsafe 创建的 Buffer 实例通常比 Buffer.alloc 创建的要快,因为它避免了初始化内存的步骤。
    • 如果你需要确保 Buffer 中的数据是干净的,你应该在使用 Buffer.allocUnsafe 创建实例后,使用 Buffer.fill() 方法来填充整个 Buffer,或者在写入数据之前完全覆盖它。
  1. Buffer.alloc(size[, fill[, encoding]]):
    • Buffer.alloc 创建一个指定大小的 Buffer 实例,并且会用指定的值(默认为0)初始化整个 Buffer。这确保了新创建的 Buffer 实例不会包含任何旧数据。
    • Buffer.alloc 方法比 Buffer.allocUnsafe 慢,因为它需要额外的时间来初始化内存。
    • 如果你不需要处理可能包含敏感数据的旧内存,或者你打算立即用新数据覆盖整个 Buffer,那么使用 Buffer.alloc 是一个更安全的选择。

在实际应用中,如果你需要处理敏感数据或者需要确保 Buffer 的内容是可预测的,建议使用 Buffer.alloc。如果你对性能有更高的要求,并且能够确保在读取或使用 Buffer 之前清除或覆盖其内容,那么可以考虑使用Buffer.allocUnsafe。

在ws源码中,使用的是Buffer.allocUnsafe,可能是为了性能优化和内存管理。在 Node.js 中,Buffer.allocUnsafe 可能会使用一个内部的内存池来分配 Buffer 实例。这意味着,如果创建的 Buffer 大小小于或等于 Buffer.poolSize 的一半,那么这些 Buffer 实例可能会共享内存池中的内存。这有助于减少内存分配和垃圾回收的开销。

当然在ws源码中,处理数据帧的函数是frame

能看到,除了负载数据之外,ws中还有很多其他参数,比如说处理掩码,比如是否设置FIN,指定是否可以修改’ data '等等。但是正常使用就是上面的encodeMessagePerf流程了。

接下来就完美解决了数据大量的问题。

如何区分不同客户端?

能看到目前只是一个客户端,那客户端一多,想要控制A客户端,发消息给B客户端,B客户端发消息给A客户端,那这样就要区分不同的客户端了。

那如何区分不同的客户端呢?

JavaScript中有一个数据结构,能够将不同的数据集区分开来,就是Set啦,看一下ws的源码

客户端存储在一个Set对象中,连接的时候,就添加到clients中,断开连接了,就delete掉,这样就能清楚有几个客户端了。

看完之后,就动手撸一下试试

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.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);socket.on('data', (data) => {this.processData(data);});socket.on('close', () => {this.clients.delete(socket)console.log('close')})socket.on('end', () => {this.clients.delete(socket)console.log('end')})socket.on('error', (err) => {console.log('error', err)})if (this.clients) {this.clients.add(socket)}// debuggerconsole.log(this.clients.size, 'clients.size')});}
}

这样子就搭建好了,看一下客户端的数量是否正确

能看到数量是正常的,然后就是将A客户端的数据发送给B客户端了

ws.on('data', function connection({realDataBuffer, clients}) {// 需要在data这个回调中拿到clients数据console.log('Client connected', realDataBuffer.toString('utf8'));// ws.send(data + ' ' + Date.now()); // 发送消息// 将消息发送给所有客户端if (clients) {clients.forEach(function each(client) {if (client !== ws.socket) {ws.send(realDataBuffer); // 客户端接受的是blob格式数据}});}
});

这里,需要在data的回调中拿到clients数据,所以需要传递数据出来。由于clients在constructor中定义的,所以可以直接在emit回调的时候,将clients传递出来。

这样,外面就能访问到clients了,然后我们直接这样传递数据,但是你以为这就完了吗?那你就太天真了🤣😂

在这里插入图片描述

找一下问题,我做了一个简单的demo,就是A页面发消息给B页面,B页面收到了,然后B页面发消息给A页面,A页面并没有收到,然后node端是都能看到AB各自发送的消息的。

其实就是node端发送消息出现了问题,仔细看一下下面的代码

ws.on('data', function connection({realDataBuffer, clients}) {console.log('Client connected', realDataBuffer.toString('utf8'));// ws.send(data + ' ' + Date.now()); // 发送消息// 将消息发送给所有客户端if (clients) {clients.forEach(function each(client) {if (client !== ws.socket) { // ws.socket是最后连接的客户端,不是当前要发消息的客户端ws.send(realDataBuffer); // 客户端接受的是blob格式数据}});}
});

原因就是ws.socket是最后连接的客户端,不是当前发生消息的客户端。

所以导致,我首先刷新的A页面,再刷新了B页面,ws.socket出现了覆盖,此时ws.socket就是B客户端了。

所以后面B页面发送消息给A页面,A页面没有收到,而是又再次将消息发送给了B页面。

如何将A的数据,广发给其他客户端?

如何解决呢?我们仔细看一下ws是如何将一个客户端的数据,发送给除了当前以外的其他客户端。

可以看到这里是先有connection,然后才能触发message回调

而且在connection回调中,拿到了ws对象,这个对象的message回调才会监听到客户端发送过来的消息

也就是说我们上面的data回调需要再包一层,然后判断是通过ws这个对象来判断是不是当前需要发送消息的客户端的。

这样就觉得ws其实就是一个客户端对象,是clients中的一个对象,然后ws的message回调,就是接收到当前客户端的消息,然后只要将不是当前ws的客户端发送消息,就可以了。

所以这里区分成服务端的socket管理和单个客户端的socket管理,需要在再新增一个ws对象,也需要继承自EventEmitter,因为它也有on事件

class SingleData extends EventEmitter  {constructor(socket) {super();this.socket = socketthis.socket.on('data', (data) => { // 传输数据this.processData(data);});this.socket.on('close', () => {this.emit('close')})this.socket.on('end', () => {this.emit('end')console.log('end')})this.socket.on('error', (err) => {this.emit('error', err)})}handleRealData(opcode, realDataBuffer) {switch (opcode) {case OPCODES.TEXT: // 文本this.emit('message', realDataBuffer);break;case OPCODES.BINARY: // 二进制this.emit('message', realDataBuffer);break;default:this.emit('close');break;}}processData(bufferData) {const byte1 = bufferData.readUInt8(0);let opcode = byte1 & 0x0f; const byte2 = bufferData.readUInt8(1);const str2 = byte2.toString(2);const MASK = str2[0];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;}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);} this.handleRealData(opcode, realData);}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.log(data)console.error('暂不支持发送的数据类型')}this.doSend(opcode, buffer);}doSend(opcode, bufferDatafer) {// 大量数据let list = frame(bufferDatafer)if (list.length === 2) {this.socket.cork();this.socket.write(list[0]);this.socket.write(list[1]);this.socket.uncork();} else {this.socket.write(list[0]);}}
}module.exports = SingleData;

由于这是单个的客户端socket数据,所以将接受和发送socket数据帧的处理,放到这个对象来。

然后就可以将服务端的socket对象,简化成下面这样

const { EventEmitter } = require('events');
const http = require('http');
const crypto = require('crypto');
const SingleData = require('./clientSocket.js')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.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);const ws = new SingleData(socket);ws.on('close', () => {this.clients.delete(ws)console.log('close')})ws.on('end', () => {this.clients.delete(ws)console.log('end')})ws.on('error', (err) => {console.log('error', err)})if (this.clients) {this.clients.add(ws)}this.emit('connection', ws);});}}module.exports = MyWebsocket;

再看看效果,这样就实现了一个简易版的ws源码啦~

在这里插入图片描述

当然ws源码中的处理,比上面这个简易版复杂很多,发送消息有Sender类,接受消息有Receiver类,实现双向通讯,而且数据处理上,也更加谨慎。

如果你想深入研究源码,可以看看这篇文章:https://juejin.cn/post/6844903850667671560

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/news/699346.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

详解AP3216C(三合一sensor: 光照、距离、照射强度)驱动开发

目录 概述 1 认识AP3216C 1.1 AP3216C特性 1.2 AP3216C内部结构 1.3 AP3216C 硬件电路 1.4 AP3216C工作时序 1.4.1 I2C 写数据协议 1.4.2 I2C 读数据协议 1.5 重要的寄存器 1.5.1 系统配置寄存器 1.5.2 和中断相关寄存器 1.5.3 IR数据寄存器 1.5.4 ALS 数据寄存器 …

代码随想录算法训练营第59天 | 583.两个字符串的删除操作 + 72.编辑距离 + 编辑距离总结篇

今日任务 583. 两个字符串的删除操作 72. 编辑距离 编辑距离总结篇 583.两个字符串的删除操作 - Medium 题目链接&#xff1a;. - 力扣&#xff08;LeetCode&#xff09; 给定两个单词 word1 和 word2 &#xff0c;返回使得 word1 和 word2 相同所需的最小步数。 每步 可以…

【数学建模规则】2024年第九届数维杯大学生数学建模挑战赛参赛指南

一、竞赛介绍 数维杯大学生数学建模挑战赛每年分为两场&#xff0c;每年上半年为数维杯国赛&#xff08;5月&#xff0c;俗称小国赛&#xff09;&#xff0c;下半年为数维杯国际赛(11月)&#xff0c;2023年第八届数维杯大学生数学建模挑战赛共有近1.4万名学生参赛&#xff0c;…

Unity之PUN2插件实现多人联机射击游戏

目录 &#x1f4d6;一、准备工作 &#x1f4fa;二、UI界面处理 &#x1f4f1;2.1 登录UI并连接PUN2服务器 &#x1f4f1;2.2 游戏大厅界面UI &#x1f4f1;2.3 创建房间UI &#x1f4f1;2.4 进入房间UI &#x1f4f1;2.5 玩家准备状态 &#x1f4f1;2.6 加载战斗场景…

vue proxy解决跨域问题

https://blog.csdn.net/tttttrrrhh/article/details/127685318 cims系统ui

java 面向对象-上:类的结构之二

类的设计中&#xff0c;两个重要结构之二&#xff1a;方法 方法 描述类应该具的功能。 比如&#xff1a;Math类&#xff1a;sqrt()\random() \... Scanner类&#xff1a;nextXxx() ... Arrays类&#xff1a;sort() \ binarySearch() \ toString() \ equals() \ ... 1.举例 p…

[Java 项目亮点] 三层限流设计

思路来源&#xff1a;bilibili 河北王校长 文章目录 面试官可能会问你能详细介绍一下Nginx的http_limit_req_module模块吗&#xff1f;你能解释一下如何在Nginx中配置http_limit_req_module模块吗&#xff1f;你知道如何调整Nginx的http_limit_req_module模块以适应不同的业务需…

【服务器数据恢复】通过reed-solomon算法恢复raid6数据的案例

服务器数据恢复环境&#xff1a; 一台网站服务器中有一组由6块磁盘组建的RAID6磁盘阵列&#xff0c;操作系统层面运行MySQL数据库和存放一些其他类型文件。 服务器故障&#xff1a; 该服务器在工作过程中&#xff0c;raid6磁盘阵列中有两块磁盘先后离线&#xff0c;不知道是管理…

Linux--自定义shell

shell shell就是操作系统提供给用户与操作系统进行交互的命令行界面。它可以理解为一个用户与操作系统之间的接口&#xff0c;用户可以通过输入命令来执行各种操作&#xff0c;如文件管理、进程控制、软件安装等。Shell还可以通过脚本编程实现自动化任务。 常见的Unix系统中使…

创新性3D数据合成模型,微软推出EgoGen

随着AR、VR等设备的广泛应用,第一人称的应用开始增多。但在研发方面面临不同的挑战,例如&#xff0c;图像模糊、视觉混乱、遮挡更严重等&#xff0c;给视觉模型的训练带来重大挑战。 一方面,人工标注真实第一视角数据集&#xff0c;来培训深度学习模型的成本和难度都很高。另一…

Java 泛型

优质博文&#xff1a;IT-BLOG-CN 一、为什么要有泛型 【1】解决元素存储的安全性问题。 【2】解决获取数据元素时&#xff0c;需要类型强转的问题。 【3】可以统一数据类型&#xff0c;便于操作。 【4】将运行时的异常提前到了编译时&#xff0c;提高了效率。 【5】实现代码的…

Day20_网络编程(软件结构,网络编程三要素,UDP网络编程,TCP网络编程)

文章目录 Day20 网络编程学习目标1 软件结构2 网络编程三要素2.1 IP地址和域名1、IP地址2、域名3、InetAddress类 2.2 端口号2.3 网络通信协议1、OSI参考模型和TCP/IP参考模型2、UDP协议3、TCP协议 2.4 Socket编程 3 UDP网络编程3.1 DatagramSocket和DatagramPacket1、Datagram…

小世界网络:直径、分形、同配性

1.小世界网络特点 —— 网络直径接近于网络中节点数量的自然对数 2.小世界分形网络 —— 移除弱链接的小世界网络 3.同配性分析 —— Pearson相关系数、邻居相关度 在宏观层面上&#xff0c;关注平均度、度分布和聚类等全局结构特征的影响。更高的平均度被认为会导致更…

Zookeeper简介及选举机制

1.概述 Zookeeper是一个开源的&#xff0c;分布式的&#xff0c;为分布式框架&#xff08;如下图中的Hadoop和Hive&#xff09;提供协调服务的Apache项目。 工作机制&#xff1a;基于观察者设计模式的分布式服务管理框架&#xff0c;负责存储和管理数据&#xff0c;接受观察者…

[算法沉淀记录] 排序算法 —— 归并排序

排序算法 —— 归并排序 算法介绍 归并排序是一种分治算法&#xff0c;由约翰冯诺伊曼在1945年发明。它的工作原理是将未排序的列表划分为n个子列表&#xff0c;每个子列表包含一个元素(包含一个元素的列表被认为是有序的)&#xff0c;然后重复合并子列表以生成新的有序子列表…

ClickHouse 指南(三)最佳实践 -- 稀疏主索引

在ClickHouse主索引的实用介绍 ClickHouse release 24.1, 2024-01-30 1、简介 在本指南中&#xff0c;我们将深入研究ClickHouse索引。我们将详细说明和讨论: ClickHouse中的索引与传统的关系数据库管理系统有何不同ClickHouse是如何构建和使用表的稀疏主索引的什么是在Clic…

JavaScript原型继承与面向对象编程思想

原型继承与面向对象编程思想 在JavaScript中&#xff0c;原型(prototype)、构造函数(constructor)和实例对象(instance)是面向对象编程中的重要概念&#xff0c;并且它们之间存在着紧密的关系。 原型(prototype)&#xff1a;原型是JavaScript中对象之间关联的一种机制。每个Ja…

js使用import到本js文件中的函数时报错 Error [ERR_MODULE_NOT_FOUND]: Cannot find module

node:internal/process/esm_loader:97internalBinding(errors).triggerUncaughtException(^Error [ERR_MODULE_NOT_FOUND]: Cannot find module D:\桌面\Pagesizedetection\lib\screensize imported from D:\桌面\Pagesizedetection\index.js Did you mean to import ../lib/sc…

网页数据的存储--存储为文本文件(TXT、JSON、CSV)

用解析器解析出数据后&#xff0c;接下来就是存储数据了。数据的存储有多种多样&#xff0c;其中最简单的一种是将数据直接保存为文本文件&#xff0c;如TXT、JSON、CSV等。这里就介绍将数据直接保存为文本文件。 目录 一、Python存储数据的方法 1、 文件读取 2、 文件写入…

【Logback】Logback 日志框架的架构

目录 1、Logger&#xff08;记录器&#xff09; &#xff08;1&#xff09;有效级别和级别继承 &#xff08;2&#xff09;日志打印和日志筛选 &#xff08;3&#xff09;记录器命名 2、Appenders&#xff08;追加器&#xff09; 3、Layouts&#xff08;布局&#xff09;…