[JavaScript知识库]5.2 使用worker_thread进行并发

1、worker_thread通过ShareArrayBuffer或者ArrayBuffer来实现共享内存

ArrayBuffer本质上是通过postMessage()内存移动的方式来使各个worker之间共享数据的，但是一旦转移后，其他的worker就不能使用，也就是同一时间，只能有个worker拿到这块内存区域

示例如下

run.js

const {Worker,  MessageChannel} = require('worker_threads')
const array = new ArrayBuffer(2)
const int32 = new Int16Array(array)
const subChannel = new MessageChannel();
const worker1 = new Worker('./worker1.js');
const worker2 = new Worker('./worker2.js');
worker1.postMessage({ hereIsYourPort: subChannel.port1,value:int32 }, [subChannel.port1,int32.buffer]);
worker2.postMessage({ hereIsYourPort2: subChannel.port2 }, [subChannel.port2]);

worker1.js

const {
    parentPort
} = require('worker_threads');

parentPort.on('message',function (msg){
    msg.value[0] = 1
    msg.hereIsYourPort.postMessage({value:msg.value},[msg.value.buffer])
    msg.hereIsYourPort.on('message',function (msg){
        console.log(msg)
    })
})

worker2.js

const {
    parentPort
} = require('worker_threads');

parentPort.on('message',function (msg){
    msg.hereIsYourPort2.on('message',function (data){
        msg.hereIsYourPort2.postMessage('receive ArrayBuffer,the value is ' + data.value[0])
    })
})

2、使用ShareArrayBuffer来实现共享，通过ShareArrayBuffer实现一个简单的并发

run.js

const {
    Worker
} = require('worker_threads');

let success = new SharedArrayBuffer(4)
let fail = new SharedArrayBuffer(4)

for (let i = 0; i < 200; i++) {
    const worker = new Worker('./worker1.js');
    worker.postMessage({'threadid': `${worker.threadId}`, 'success': success, 'failed': fail})
    worker.on('message', function (msg) {
        if (msg.msg === 'exit') {
            console.log(worker.threadId + ":线程结束")
            console.log("成功:" + msg.pass)
            console.log("失败:" + msg.fail)
            worker.terminate()
        }
    })
}

worker1.js

const {
    parentPort
} = require('worker_threads');
const axios = require('axios');

parentPort.on('message', async function (msg) {
    let pass = msg.success;
    let passCount = new Int32Array(pass)
    let fail = msg.failed
    let failCount = new Int32Array(fail)
    for (let i = 0; i < 200; i++) {
        const data = {
            name: msg.threadid + ':' + i
        }
        await axios.post('http://127.0.0.1:9898/api/category/create', data, {headers: {'cookie': 'EGG_SESS=2sTjpvl0ktiRxnHwosRRCFEKYuyBVYDEOUV4imobh9spjTflWDF1zz-BEh7DZ4nG'}}).then(res => {
            if (res.data._id !== undefined) {
                Atomics.add(passCount, 0, 1)
            } else {
                Atomics.sub(failCount, 0, 1)
            }
        }).catch(error => {
            Atomics.sub(failCount, 0, 1)
        })
    }
    parentPort.postMessage({'msg': 'exit', 'pass': Atomics.load(passCount, 0), 'fail': Atomics.load(failCount, 0)})
})

验证结果，失败23058个，成功16942个

通过db可以看到，发送的40000个请求，成功数与预期数是一致的

?

?Atomics提供了针对ShareArrayBuffer中的数据原子性操作

Atomics文档：https://developer.mozilla.org/zh-CN/docs/Web/JavaScript/Reference/Global_Objects/Atomics/add

TypedArray文档：

https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Global_Objects/TypedArray?

TypedArray的使用和DataView的使用很类似