【每日一题】- 2020-08-03 - 多线程打印

[azl397985856]

通过2个线程，交替输出 A1B2C3....Z26 。

比如：

• 线程 1 打印 A
• 线程2 打印 1
• 。。。

[suukii]

Node

// thread1.js
const { Worker } = require('worker_threads');
const worker = new Worker('./thread2.js', {});

let code = 'A'.charCodeAt(0);
worker.on('message', msg => {
    process.stdout.write(msg);
    code++;
    code <= 'Z'.charCodeAt(0) && worker.postMessage(String.fromCharCode(code));
});

worker.postMessage(String.fromCharCode(code));

// thread2.js
const { parentPort } = require('worker_threads');

let count = 1;
parentPort.on('message', msg => {
    process.stdout.write(msg);
    count <= 26 && parentPort.postMessage(String(count));
    count++;
});

[azl397985856]

@suukii 如何保证两个线程交替执行呢？

[feikerwu]

@suukii 如何保证两个线程交替执行呢？

自实现一个线程池，限制并发，通过message事件循环

[suukii]

好像这样可以吧，ps 为了让代码短一点 TestWorkerPool 的实现我就尽量写简单了。

不过为什么要让线程交替执行呢

const {
    Worker,
    isMainThread,
    parentPort,
    workerData,
} = require('worker_threads');

class TestWorkerPool {
    constructor(workerPath, numOfThreads) {
        this._workerPath = workerPath;
        this._queue = [];

        this._numOfThreads = numOfThreads;
        this._workersById = Array(numOfThreads);
        this._nextWorkerId = -1;
        this._activeWorker = null;

        this._init();
    }

    _init() {
        for (let i = 0; i < this._numOfThreads; i++) {
            this._workersById[i] = new Worker(this._workerPath, {
                workerData: { id: i },
            });
        }
    }

    _getNextWorkerId() {
        this._nextWorkerId++;
        if (this._nextWorkerId >= this._numOfThreads) this._nextWorkerId = 0;
        return this._nextWorkerId;
    }

    _runWorker(workerId, queueItem) {
        const worker = this._workersById[workerId];
        this._activeWorker = worker;

        const cleanUp = () => {
            worker.removeAllListeners('message');
            worker.removeAllListeners('error');
            this._activeWorker = null;

            if (this._queue.length)
                this._runWorker(this._getNextWorkerId(), this._queue.shift());
        };

        worker.once('message', cleanUp);
        worker.once('error', cleanUp);

        worker.postMessage(queueItem);
    }

    run(data) {
        const queueItem = data;
        if (this._activeWorker) {
            this._queue.push(queueItem);
            return null;
        }
        this._runWorker(this._getNextWorkerId(), queueItem);
    }
}

if (isMainThread) {
    const pool = new TestWorkerPool('./test.js', 2);
    for (let i = 1; i <= 26; i++) {
        pool.run(String.fromCharCode(i + 64));
        pool.run(String(i));
    }
} else {
    parentPort.on('message', data => {
        console.log(`worker${workerData.id}: ${data}`);
        parentPort.postMessage('done');
    });
}

// worker0: A;
// worker1: 1;
// worker0: B;
// worker1: 2;
// worker0: C;
// worker1: 3;

[stale[bot]]

This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs. Thank you for your contributions.