node之Buffer--以websocket不完全实现为例

HTKLPXH

2019-06-28

关注关注

Buffer

定义

Buffer是node提供的一个操作二进制流的类。它能够让开发者以类似操作数组的方式去操作二进制数据。事实上，es6也提供了定型数组来操作二进制数据，后文会简单对比下两者区别。

Buffer的使用场景

回顾二进制操作

假设

a=0001 0101(2)
b=0101 1010(2)

1.| 位或, 对应位有一个为1则结果为1

a | b = 0101 1111(2)

2.& 位与，对应位都为1结果才为1

a & b = 0001 0000(2)

3.~ 位非, 对应位1变0，0变1

~a = 1110 1010(2)

console.log('~a',(~a).toString(2));    
js中运行结果-10110 也就是1001 0110;
这个过程可能是，求位非后得到111 01010，发现是个负数，存它的补码即1001 0110，读到的就是这个值了

4.>> 向右移位所有位像左移指定的位数，左边超出的舍弃，右边不足的补0

a >> 2 = 0000 0101(2)

5.<< 向左移位所有位像左移指定的位数，左边超出的舍弃，右边不足的补0

a << 2 = 0101 0100(2)

java 里有带符号右移和带符号左移，js里是没有这种操作的

6.^ 异或对应位相同则为0，否则为1

a ^ b = 01001111

7.一些操作

取出a的低4位

a & 0xf

取出a的高4位

a >> 4

取出a的高第3位

(a & 0x20) >> 5

其他的方式自己探索去吧
关于位运算，可以参看https://blog.csdn.net/foreverling_ling/article/details/61417649

Buffer的创建

node v6.0之前可以用new Buffer()创建，以后则用如下几个方法创建

Buffer.alloc(size[, fill[, encoding]])
Buffer.allocUnsafe(size)
Buffer.allocUnsafeSlow(size)
Buffer.from(array) // array 必须是八位字节数组，否则有问题
Buffer.from(arrayBuffer[, byteOffset[, length]])
Buffer.from(buffer)
Buffer.from(string[, encoding])

关于这些方法的具体用法可参考文档，这里简单演示一些基础用法. 需要注意的是，
Buffer中的一项是一个字节（8位）。log出来每个项用两个16进制数表示。

let buf1 = Buffer.alloc(8);
let buf2 = Buffer.alloc(8,1);
let buf3 = Buffer.alloc(8,'a');
let buf4 = Buffer.alloc(8, 'a', 'utf8');

let buf5 = Buffer.allocUnsafe(8);
let buf6 = Buffer.allocUnsafeSlow(8);


console.log('buf1', buf1);  // buf1 <Buffer 00 00 00 00 00 00 00 00>
console.log('buf2', buf2);  // buf2 <Buffer 01 01 01 01 01 01 01 01>
console.log('buf3', buf3);  // buf3 <Buffer 61 61 61 61 61 61 61 61>
console.log('buf4', buf4);  // buf4 <Buffer 61 61 61 61 61 61 61 61>


console.log('buf5', buf5);  // buf5 <Buffer c0 1d e0 03 01 00 00 00>
console.log('buf6', buf6);  // buf6 <Buffer 00 00 00 00 00 00 00 00>

// 注意buf6与buf1对比


let buf7 = Buffer.from([10,0x61,0b10,'a']);
console.log('buf7', buf7);  // buf7 <Buffer 0a 61 02 00>

let arrayBuffer = new Uint16Array(2);
arrayBuffer[0] = 10;
arrayBuffer[1] = 12;
console.log('arrayBuffer.byteLength', arrayBuffer.byteLength);  // arrayBuffer.byteLength 4

// 这样写是不共享内存的, 新开辟了一块内存, 而且去掉了为0的字节
let buf8 = Buffer.from(arrayBuffer);
console.log('buf8', buf8);  // buf8 <Buffer 0a 0c>
arrayBuffer[0] = 11;
console.log('buf8', buf8);  // buf8 <Buffer 0a 0c>

// 这样做跟ArrayBuffer是共享内存的
let buf9 = Buffer.from(arrayBuffer.buffer);
console.log('buf9', buf9);  // buf9 <Buffer 0b 00 0c 00>

arrayBuffer[1] = 14;
console.log('buf9', buf9);  // buf9 <Buffer 0b 00 0e 00>

let buf10 = Buffer.alloc(8, 1);
// 从Buffer创建Buffer也是会新开辟内存的, 不共享
let buf11 = Buffer.from(buf10);

console.log('buf11', buf11);    // buf11 <Buffer 01 01 01 01 01 01 01 01>
buf10[0] = 2;
console.log('buf10', buf10);    // buf10 <Buffer 02 01 01 01 01 01 01 01>
console.log('buf11', buf11);    // buf11 <Buffer 01 01 01 01 01 01 01 01>

读Buffer

buf.readDoubleBE(offset)
buf.readDoubleLE(offset)
buf.readFloatBE(offset)
buf.readFloatLE(offset)
buf.readInt8(offset)
buf.readInt16BE(offset)
buf.readInt16LE(offset)
buf.readInt32BE(offset)
buf.readInt32LE(offset)
buf.readIntBE(offset, byteLength)
buf.readIntLE(offset, byteLength)
buf.readUInt8(offset)
buf.readUInt16BE(offset)
buf.readUInt16LE(offset)
buf.readUInt32BE(offset)
buf.readUInt32LE(offset)
buf.readUIntBE(offset, byteLength)
buf.readUIntLE(offset, byteLength)

具体用法可参考文档，这里比较下有U和无U， BE和LE

1.BE和LE

BE是按大端序去读
认为低地址存的是高位
LE是按小端序去读
相反，认为低地址存的是低位

大端序用的多一点，符合习惯

let buf = Buffer.from([0x01, 0x02, 0x03, 0x01, 0x02]);

 console.log(buf.readInt16BE(0));   // 258
 console.log(buf.readInt16LE(0));   // 513

 console.log(0x0102); // 258
 console.log(0x0201); // 513

2.有U和无U

主要是看第一位算不算符号位，无U的话第一位是算符号位的，有U第一位算数据位

let buf2 = Buffer.from([0xff]);
 console.log(buf2.readInt8(0));     // -1
 console.log(buf2.readUInt8(0));    // 255

注意下上边的offset是偏移多少字节，单位字节

写Buffer

buf.write(string[, offset[, length]][, encoding])
buf.writeDoubleBE(value, offset)
buf.writeDoubleLE(value, offset)
buf.writeFloatBE(value, offset)
buf.writeFloatLE(value, offset)
buf.writeInt8(value, offset)
buf.writeInt16BE(value, offset)
buf.writeInt16LE(value, offset)
buf.writeInt32BE(value, offset)
buf.writeInt32LE(value, offset)
buf.writeIntBE(value, offset, byteLength)
buf.writeIntLE(value, offset, byteLength)
buf.writeUInt8(value, offset)
buf.writeUInt16BE(value, offset)
buf.writeUInt16LE(value, offset)
buf.writeUInt32BE(value, offset)
buf.writeUInt32LE(value, offset)
buf.writeUIntBE(value, offset, byteLength)
buf.writeUIntLE(value, offset, byteLength)

与读操作类似，不做演示

类数组操作

1.访问和写入

buf[0] // 访问第一个字节
buf[0] = 1 // 将第一个字节写为1

2.slice方法

从一个buffer中截取一段生成一个buffer，新旧buffer是共享内存的，和Array不同，Array的这个方法会新开辟内存。

let buf = Buffer.from([0x01, 0x02, 0x03]);

console.log(buf);   // <Buffer 01 02 03>

buf[0] = 0x5;
console.log(buf[0]);    // 5

let buf2 = buf.slice(0,2);
console.log(buf2);      // <Buffer 05 02>
buf[1] = 0x8;
console.log(buf2);      // <Buffer 05 08>


let arr = [0,1,2,3];
let arr2 = arr.slice(0,2);
arr[0] = 10;
console.log(arr2);  // [0, 1]

3.concat方法

连接多个buffer生成一个buffer，不共享内存

let buf3 = Buffer.alloc(4,1);
let buf4 = Buffer.alloc(4,5);
let totalLength = buf3.length + buf4.length;

const bufA = Buffer.concat([buf3, buf4], totalLength);

console.log(bufA);  // <Buffer 01 01 01 01 05 05 05 05>
buf3[0] = 3;
console.log(bufA);  // <Buffer 01 01 01 01 05 05 05 05>

关于Buffer更深入的理解可以参考

https://segmentfault.com/a/1190000005368752
https://segmentfault.com/a/1190000008772711

ArrayBuffer

定义

ArrayBuffer在ES6中被引入，可以认为是一定大小的内存空间。对这段内存的操作则需要视图支持，包括通用的视图DataView和定型数组1TypedArray。

ArrayBuffer的基本操作

ArrayBuffer(size: number): ArrayBuffer // 创建size个字节的ArrayBuffer
slice(start: number, end: number): ArrayBuffer // 截取生成新的ArrayBuffer，不共享内存

let buffer1 = new ArrayBuffer(8);

let view1 = new DataView(buffer1);
console.log(buffer1);   // ArrayBuffer { byteLength: 8 }
console.log(view1);
// DataView {
//     byteLength: 8,
//     byteOffset: 0,
//     buffer: ArrayBuffer { byteLength: 8 } }

view1.setInt8(0,3);
console.log(view1.getInt8(0));  // 3

let buffer2 = buffer1.slice(1,2);
let view2 = new DataView(buffer2);
view1.setInt8(1,3);
console.log(view2.getInt8(0));  // 0

DataView

1.属性

buffer
byteLength
byteOffset

2.写操作

setInt8(byteOffset: number, value: number, [isLittleEndian]): void
setUInt8(byteOffset: number, value: number, [isLittleEndian]): void
setInt16(byteOffset: number, value: number, [isLittleEndian]): void
setUInt16(byteOffset: number, value: number, [isLittleEndian]): void
setInt32(byteOffset: number, value: number, [isLittleEndian]): void
setUInt32(byteOffset: number, value: number, [isLittleEndian]): void
setFloat32(byteOffset: number, value: number, [isLittleEndian]): void
setFloat64(byteOffset: number, value: number, [isLittleEndian]): void

3.读操作

getInt8(byteOffset: number, value: number, [isLittleEndian]): void
getUInt8(byteOffset: number, value: number, [isLittleEndian]): void
getInt16(byteOffset: number, value: number, [isLittleEndian]): void
getUInt16(byteOffset: number, value: number, [isLittleEndian]): void
getInt32(byteOffset: number, value: number, [isLittleEndian]): void
getUInt32(byteOffset: number, value: number, [isLittleEndian]): void
getFloat32(byteOffset: number, value: number, [isLittleEndian]): void
getFloat64(byteOffset: number, value: number, [isLittleEndian]): void

TypedArray

1.类型

Int8Array
Uint8Array
Uint8ClampedArray
Int16Array
Uint16Array
Int32Array
Uint32Array
Float32Array
Float64Array

创建方法

new TypedArray(buf: Buffer):TypedArray
new TypedArray(size: number):TypedArray
new TypedArray(typedArray: TypedArray): TypedArray
new TypedArray(array: Array): TypedArray

2.方法

typedArray[0] = 1
console.log(typedArray[0])

属性

byteLength: 占据的字节数
length: 元素个数
buffer: 使用的ArrayBuffer

let buffer1 = new ArrayBuffer(10);

let int8Array = new Int8Array(buffer1);

int8Array[0] = 1;

console.log(int8Array.length);          // 10
console.log(int8Array.byteLength);      // 10
console.log(int8Array[0]);              // 1

let int16Array = new Int16Array(buffer1);
console.log(int16Array.length);         // 5
console.log(int16Array.byteLength);     // 10

node Buffer, TypedArray, Array比较

Array是不能去操作位的，但是Buffer和TypedArray是可以的
Buffer的slice方法是在原来的内存上直接截取的，会共享一段内存，而Array和TypedArray是会拷贝一份放到一段新的内存
Buffer是Uint8Array的实例，buf instanceOf Unint8Array=true
Buffer和Array都没有byteLength方法，而TypedArray有，byteLength表示TypedArray占用的字节数，length表示的是有多少个项
Array数据项没有一定的数据类型，Buffer的每一项都是1字节数字，TypedArray有多种类型

Array的文档：
https://developer.mozilla.org/zh-CN/docs/Web/JavaScript/Reference/Global_Objects/Array/copyWithin
TypedArray文档：
https://developer.mozilla.org/zh-CN/docs/Web/JavaScript/Reference/Global_Objects/TypedArray
Buffer文档
http://nodejs.cn/api/buffer.html

websocket协议的简单实现

协议解读

1.介绍

WebSocket协议是基于TCP的一种新的网络协议。它实现了浏览器与服务器全双工(full-duplex)通信——允许服务器主动发送信息给客户端。

2.连接（握手）

客户端 -> 发一个特殊的http请求 -> 服务器发一个特殊的http响应 -> 连接成功

连接请求

下面是个实际的请求头

GET ws://127.0.0.1:3000/ HTTP/1.1
Host: 127.0.0.1:3000
Connection: Upgrade
Pragma: no-cache
Cache-Control: no-cache
User-Agent: Mozilla/5.0 (Macintosh; Intel Mac OS X 10_13_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/69.0.3497.100 Safari/537.36
Upgrade: websocket
Origin: http://localhost:3000
Sec-WebSocket-Version: 13
Accept-Encoding: gzip, deflate, br
Accept-Language: zh-CN,zh;q=0.9
Sec-WebSocket-Key: HfqW8RyI8GitR89fzjbGgA==
Sec-WebSocket-Extensions: permessage-deflate; client_max_window_bits

一个普通的请求头如下

GET / HTTP/1.1
Host: localhost:3000
Connection: keep-alive
Pragma: no-cache
Cache-Control: no-cache
Upgrade-Insecure-Requests: 1
User-Agent: Mozilla/5.0 (Macintosh; Intel Mac OS X 10_13_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/69.0.3497.100 Safari/537.36
Accept: text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8
Accept-Encoding: gzip, deflate, br
Accept-Language: zh-CN,zh;q=0.9

关于建立连接的请求不详细说，重点关注的服务端的部分，使用h5的websocket时，浏览器已经帮我们做好了一切。想了解更多细节的同学可以参考

连接响应

下面是一个实际的响应

HTTP/1.1 101 Switching Protocols
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Accept: mG8+Ke3Gs4TDeff7HYfHmoXPkrA=

在node中，使用createServer会返回一个http.Server类的实例，当客户端对这个server发起一个Connection为Upgrade的请求时，server会触发一个Upgrade事件，我们可以在这个事件处理函数中多客户端发来的信息进行校验，校验成功可以按协议给出响应即建立连接成功了。

const crypto = require('crypto');
const GUID = '258EAFA5-E914-47DA-95CA-C5AB0DC85B11';  // GUID
server.on('upgrade', (req, socket) => {
    let key = req.headers['sec-websocket-key'];
    key = crypto
        .createHash('sha1')
        .update(key + GUID)
        .digest('base64');
    let resHeaders = [
        'HTTP/1.1 101 Switching Protocols',
        'Upgrade: websocket',
        'Connection: Upgrade',
        'Sec-WebSocket-Accept: ' + key,
        '',
        ''
    ];
    resHeaders = resHeaders.join('\r\n');

    socket.write(resHeaders);
});

上述代码忽略了对请求头字段的校验。

Sec-WebSocket-Accept这个字段是需要计算的。

需要将请求头的sec-websocket-key字段与一个规定的字符串连接，再使用sha1加密，再将这个加密后的字符串转成base64即可得到。

3.帧结构

node之Buffer--以websocket不完全实现为例

FIN: 1bit

一个数据块可能被拆成多个数据帧发送, FIN标示是否为最后一个数据帧, 只有一个数据帧也要置为1

RSV1, RSV2, RSV3 1bit x 3

除非两端协商了非0值的含义, 否则必须都为0; 不然另一端要中断连接

opcode 4bits

定义数据部分的信息, 无效终止连接; 0x8标示是个关闭连接的帧

mask 1bit

标示数据部分是否使用掩码

payload length: 7bits 7+16 7+64
0-125: 7bits
126 2B
127 8B
masking-key: 0 | 4B

从客户端发送的数据会包含这个掩码

extension data + application data: x+y

x= 0B or xB, 除非已经和客户端确定扩展数据含义, 否则不能有

y 应用数据

4.断开

略了

代码实现

// 这里边还有很多问题
const EventEmitter = require('events');
const GUID = '258EAFA5-E914-47DA-95CA-C5AB0DC85B11';  // GUID
class Websocket extends EventEmitter {

    constructor(p) {
        super();
        if(typeof p === 'number') { // 表示不是挂在http服务上的
            // 略了, 可以用net 模块根据指定端口创建tcp连接
        }else if(typeof p === 'object') {   // 表示与http服务公用一个服务
            this.server = p;
            this._init();
        } else {
            throw new Error('error');
        }
        this.socketsMap = new Map();
    }
    _handshake(req, socket) {
        this.socket = socket;
        let key = req.headers['sec-websocket-key'];
        key = require('crypto')
            .createHash('sha1')
            .update(key + GUID)
            .digest('base64');
        let resHeaders = [
            'HTTP/1.1 101 Switching Protocols',
            'Upgrade: websocket',
            'Connection: Upgrade',
            'Sec-WebSocket-Accept: ' + key,
            '',
            ''
        ];
        socket.on('data', frame => {   
            const decodedFrame = this._decode(frame);
            if(decodedFrame.opcode === 0x0) {

            } 
            const opcode = decodedFrame.opcode;
            if(opcode === 0x8) {
                this.emit('close', socket);
                socket.write(this.closeFrame());
                this.socketsMap.delete(socket);
            } else {
                decodedFrame.data && this.emit('message', socket, decodedFrame.data.toString('utf8'));
            }
            
        });
        resHeaders = resHeaders.join('\r\n');
        socket.write(resHeaders);
        this.socketsMap.set(socket, 1);
        this.emit('open', socket);
    }

    _init() {
        this.server.on('upgrade', (req, socket) => {
           this._handshake(req, socket);
        })
    }

    send(socket, msg) {
        try {
            socket.write(this._encode(msg));
        } catch(e) {
            console.log(e);
        }
        
    }

    broadcast(msg) {
        for (const socket of this.socketsMap) {
            this.send(socket[0], msg);
        }
    }
    /**
     * 解码
     */
    _decode(frame) {
        console.log(frame)
        let frame1 = frame[0];  // 第一个字节
        let FIN = frame1 >> 7;  // 标示是否为结束帧
        // 扩展
        let RSV1 = frame1 >> 6 & 0b01;
        let RSV2 = frame1 >> 5 & 0b001;
        let RSV3 = frame1 >> 4 & 0b0001;
        let opcode = frame1 & 0x0F;   // 标示数据信息类型
        let MASKING_KEY_buf;    // 掩码
        let data;    // 数据
        let frame2 = frame[1];  // 第二个字节
        let MASK = frame2 >> 7;
        let payloadLength = frame2 & 0x7F;
        console.log(payloadLength)
        let extendPayloadBytes = 0;
        if (payloadLength === 126) {
            payloadLength = frame.readUInt16BE(2);
            extendPayloadBytes = 2;
        } else if (payloadLength === 127) {
            payloadLength = frame.readUInt32BE(2);
            extendPayloadBytes = 8;
        }

        if (MASK === 1) {
            MASKING_KEY_buf = frame.slice(2 + extendPayloadBytes, 6 + extendPayloadBytes);
            data = Buffer.alloc(payloadLength);
            for (let i = 0; i < payloadLength; i++) {
                var j = i % 4;
                data[i] = frame[2 + extendPayloadBytes + 4 + i] ^ MASKING_KEY_buf[j];
            }
        }

        return {
            FIN,
            RSV1,
            RSV2,
            RSV3,
            opcode,
            MASKING_KEY_buf,
            data
        }
    }

    closeFrame() {
        let f = Buffer.from([0x8, 0x8]);
        console.log(f);
        return f;
    }

    _encode(data) {
        let dataBuf = Buffer.from(data, 'utf8');
        let dataLength = dataBuf.length; // 数据长度,bytes
        let frames = [];
        let preInfoArr = [];
        preInfoArr.push((1 << 7) + 1);   // FIN和opcode
        

        if (dataLength < 126) {
            preInfoArr.push((0 << 7) + dataLength); // mask 和数据长度
            let f = Buffer.from(preInfoArr);
            dataBuf = Buffer.concat([f,dataBuf]);
            frames.push(dataBuf);   // 数据
            dataLength = 0;
        } else if (dataLength < Math.pow(2, 16)) {
            preInfoArr.push((0 << 7) + 126);    // 占位,表示要用后面两个字节标示长度
            preInfoArr.push(((dataLength & 0xFF00) >> 8), (dataLength & 0xFF));
            let f = Buffer.from(preInfoArr);
            dataBuf = f.concat(dataBuf)
            frames.push(dataBuf);
        } else if (dataLength < Math.pow(2, 32)) {
            preInfoArr.push((0 << 7) + 126);    // 占位,表示要用后面两个字节标示长度
            preInfoArr.push(0x0,0x0);
            preInfoArr.push(
                (dataLength & 0xFF000000) >> 24,
                (dataLength & 0xFF0000) >> 16,
                (dataLength & 0xFF00) >> 8,
                dataLength & 0xFF
            )
            let f = Buffer.from(preInfoArr);
            dataBuf = Buffer.concat([f,dataBuf]);            
            frames.push(dataBuf);
        } else { // 需要分片了
            // 暂不考虑了
        }
        return frames[0];
    }
}

module.exports = Websocket;

websocket