Clojure 风格的 JavaScript 并发编程

并发可能很容易和并行混淆, 但是结合我们自己想一想,还是很容易分得清的.

如果我正在上班写代码,想加个班然后发个短信给老婆说晚点回, 发完以后继续敲代码. 那么发短信和敲代码两个任务就是 并发.

但如果我还特别喜欢音乐, 所以我边听音乐边敲代码, 那么交代吗和听音乐两个任务就是并行了.

所以说, 并行与并发的最大区别就是后者任务之间是互相阻塞的, 任务不能同时运行,因此在执行一个任务时就得阻塞另外一个任务.

说到并发, 大概都会联想到多线程.

继续敲代码这个例子, 我现在fork出来一个手发短信, 但是我还是只有一个脑袋, 在发短信的时候我的脑子还是只能集中在 如何编制一个理由向老婆请假, 而另外两只手只能放在键盘上什么也改不了, 直到短信发出去, 再继续写代码.

所以多线程开销还是很大(我得再长一个手…完了还要缩回去…), 而且其他两只手其实是闲置(阻塞)着的.

因此, 另外一种更省资源的处理并发的方式就出现了–异步. 对了, 就是我们在js里经常发ajax的那个异步.

比如我还是两只手, 我发完短信继续就敲代码了, 这时, 老婆给我回了一条短信, 那我放下手中的活, 拿起手机看看居然说 同意, 于是放下手机继续敲代码了.

注意这段动作与之前多线程的区别, 多线程的场景是我fork了第三只手, 而那只手在我敲代码是一直握着手机, 等待着老婆的回复. 于是异步是不是比多线程的情况少用了只胳膊而且利用率更高呢.

那么你就要问了, 你是怎么知道手机响的, 还不是要开一个线程让耳朵监听着. 对的, 但是异步只需要很少的有限个线程就好了, 比如我有十个手机 要发给十个老婆, 我还是两个线程, 而如果是多线程的话我要fork出来十只手. JS的异步就是这么干的, 一个专门 的 event loop 用于挂各种需要执行的任务.

ES6 终于支持了Generator, 目前Firefox与Chrome都已经实现. Chrome有一个 feature toggle 可以打开部分 es6 功能. 打开 chrome://flags/#enable-javascript-harmony 设置为 true Generator在每次被调用时放回 yield 的值, 并保存状态, 下次调用时继续运行. 这种功能听起来刚好符合上例中神奇的 parking 的行为, 因此完全可以用 generator 来实现 CSP.

goroutines 其实就是一个状态机, generator为输入

• 一个函数
• 他可以接受一个 generator
• 如果generator没有下一步,则结束
• 如果该步的返回值状态为 park, 那么就是什么也不做, 过一会再来进入状态机尝试
• 如果为 continue, 这接着generator下一步, 继续循环

function go_(machine, step) {
  while(!step.done) {
    var arr   = step.value(),
        state = arr[0],
        value = arr[1];
    switch (state) {
      case "park":
        setTimeout(function() { go_(machine, step); },0);
        return;
      case "continue":
        step = machine.next(value);
        break;
    }
  }
}

function go(machine) {
  var gen = machine();
  go_(gen, gen.next());
}

一个类似于 thread sleep 的功能, 想让任务能等待个一段时间再执行, 只需要在 go_ 中加入一个 timeout 的 case 就好了

...
  case 'timeout':
    setTimeout(function(){ go_(machine, machine.next());}, value);
    return;
...

如果状态是timeout, 那么等待 value 那么长的时间再执行generator下一步.

另外还需要一个返回 timeout channel 的函数

function timeout(interval){
  var chan = [interval];
  chan.name = 'timeout';
  return chan;
}

当 generator 从 channel 中 take 数据时

• 如果 channel 空, 状态变为 park
• 如果 channel 非空, 获得数据, 状态改成 continue
• 如果是 timeout channel, 状态置成 timeout

function take(chan) {
  return function() {
    if(chan.name === 'timeout'){
      return ['timeout', chan.pop()];
    }else if(chan.length === 0) {
      return ["park", null];
    } else {
      var val = chan.pop();
      return ["continue", val];
    }
  };
}

当 generator 往 channel 中 put 数据

• 如果 channel 空, 状态变为 continue, 放入数据
• 如果 channel 非空, parking

function put(chan, val) {
  return function() {
    if(chan.length === 0) {
      chan.unshift(val);
      return ["continue", null];
    } else {
      return ["park", null];
    }
  };
}

现在可以原原本本的将之前的clojure的例子翻译成JavaScript了

function boss_yelling(){
  go(function*(){
    for(var i=0;;i++){
      yield take(timeout(1000));
      yield put(work, "boss say: work "+i);
    }
  });
}

function wife_texting(){
  go(function*(){
    for(;;){
      yield take(timeout(4000));
      yield put(text, "wife say: come home");
    }
  });
}

function working(){
  go(function*(){
    for(;;){
      var task = yield take(work);
      console.log(task, "me working");
    }
  });
}

function texting(){
  go(function*(){
    for(;;){
      var read = yield take(text);
      console.log(read, "me ignoring");
    }
  });
}
boss_yelling();
wife_texting();
working();
texting();

完整代码