【JavaScript】吃饱了撑的系列之JavaScript模拟多线程并发 - 外婆的彭湖湾

https://github.com/penghuwan/concurrent-thread.js​github.com

// 备注：为循序渐进介绍，以下为简化代码
// 存储每个线程函数的状态，例如是否中断，以及线程状态等
const threadMap = {};

class ThreadPool {
    // 模拟线程中断
    interrupt(threadName) {   }
    // 模拟线程同步
    join(threadName, targetThread) {   }
    // 模拟线程休眠
    sleep(ms) { }
};
function submit(func, name) {
    if (!func instanceof Function) return;
    // 方式1：传入一个具名函数；方式2：传入第二个参数，即线程命名空间
    const threadName = func.name || name;
    // threadMap负责存储线程状态数据
    threadMap[threadName] = { state: RUNNABLE, isInterrupted: false };
    // 让func异步调用，同时将传入函数的作用域绑定为 ThreadPool原型
    Promise.resolve({
        then: func.bind(ThreadPool.prototype);
   })
}

submit(async function example() {
    this.interrupt();
});

submit(async function example() {
    this.interrupt('example');
});

// 返回代理后的ThreadPool
function delegateThreadPool(threadName) { // threadName为待定的线程名，在submit方法调用时候传入
    // 代理后的ThreadPool
    const proxyClass = {};
    // 获取ThreadPool原来的所有的方法,赋给props数组
    var props = Object.getOwnPropertyNames(ThreadPool.prototype);
    for (let prop of props) {
        // 代理ThreadPool，为其所有方法增加threadName这个参数
        let fnName = prop;
        proxyClass[fnName] = (...args) => {
            const fn = baseClass[fnName];
            return fn(threadName, ...args);
        };
    }
    return proxyClass;
}
function submit(func, name) {
    // 省略其他代码 。。。
    const proxyScope = delegateThreadPool(threadName);
    // 让func异步调用，不阻塞主线程，同时实现并发
    Promise.resolve({
        then: function () {
            // 给func绑定this为代理后的ThreadPool对象，以便调用方法
            func.call(proxyScope);
        }
    });
}
// 调用this.sleep方法时，已经无需增加函数命名作为参数了
submit(async function example() {
    this.interrupt();
});

// 模拟“线程”休眠
sleep(ms) {
  return new Promise(function (resolve) {
    setTimeout(resolve, ms);
  })
}
// 提交“线程”
submit(async function example() {
    // 阻塞停留3秒，然后才输出1
    await this.sleep(3000);
    console.log(1);
});

    // 模拟线程中断
    interrupt(threadName) {
        if (!threadName) { throw new Error('Miss function parameters') }
        if (threadMap[threadName]) {
            threadMap[threadName].isInterrupted = true;
        }
    }
    // 获取线程中断状态
    isInterrupted(threadName) {
        if (!threadName) { throw new Error('Miss function parameters') }
        // !!的作用是：将undefined转为false
        return !!threadMap[threadName].isInterrupted;
    }

import ee from 'event-emitter';
const emitter = ee();
// 模拟线程同步
join(threadName, targetThread) {
  return new Promise((resolve) => {
    // 监听其他线程函数的结束事件
    emitter.on('join-finished', (finishThread) => {
      // 根据结束线程的线程名finishThread做判断
      if (finishThread === targetThread) {
        resolve();
      }
    })
  })
}

import ee from 'event-emitter';
const emitter = ee();
function submit(func, name) {
   // ...
    Promise.resolve({
        then: func().then(() => {
          emitter.emit('join-finished', threadName);
        })
    });
}

// 这是一个非公平锁
class Lock {
    constructor() {
        this.isLock = false;
    }
    //加锁
    lock() {
        if (this.isLock) {
            const self = this;
            // 循环while死循环，不停测试isLock是否等于false
            return new Promise((resolve) => {
                (function recursion() {
                    if (!self.isLock) {
                        // 占用锁
                        self.isLock = true;
                        // 使外部await语句继续往下执行
                        resolve();
                        return;
                    }
                    setTimeout(recursion, 100);
                })();
            });
        } else {
            this.isLock = true;
            return Promise.resolve();
        }
    }
    // 解锁
    unLock() {
        this.isLock = false;
    }
}
const lockObj = new Lock();
export default lockObj;

async function commonCode() {
    await Lock.lock();
    await Executor.sleep(3000);
    Lock.unLock();
}

submit(async function example1() {
    console.log('example1 start')
    await commonCode();
    console.log('example1 end')
});
submit(async function example2() {
    console.log('example2 start')
    await commonCode();
    console.log('example2 end')
});

async function testCode() {
    console.log('i will be wait');
    if (true) {
        await Condition.wait();
    };
    console.log('i was notified ');
}

submit(async function example() {
    testCode();
    setTimeout(() => {
        Condition.notify();
    }, 3000);
});

i will be wait
// 3秒后输出
i was notified