关于异步

• 一篇异步的文章转载 & 记录

• 资料来源:

  https://mp.weixin.qq.com/s/q6BfOINeqgm5nffrHu4kQA

• 更新

  2023.07.27 初始

导语

接下来的工作很大的重点是异步编程,协程什么的…而这些概念吧..😶‍🌫️.. 这一篇文章解答了不少疑问,感谢原作者!

正文

异步不是让单个任务执行更快,而是让计算机在相同时间内完成更多的任务.

来几段伪代码

let a = read("qq.com");
let b = read("jd.com");
print(a+b);

• read 阻塞,连续 2 个,执行时间要 x2

let op_a = read_async("qq.com");
let op_b = read_async("jd.com");
let a = “”;
let b = “”;
while true {
  if op_a.is_finish() {
    a = op_a.get_content();
    break;
  }
}
while true {
  if op_b.is_finish() {
    b = op_b.get_content();
    break;
  }
}
print(a+b);

• 先后启动读取 非阻塞; 轮询结果;
• 执行时间几乎只需 1轮等待,但 轮询 cpu 占用上去了.

读取结果就不就绪,操作系统是知道的,因此假设有这样的系统调用: wait_until_get_ready,挂起直到 io 读取就绪. 轮询的代码就能改成下面的样子

let op_a = read_async("qq.com");
let op_b = read_async("jd.com");
let a = wait_until_get_ready(op_a);
let b = wait_until_get_ready(op_b);
print(a+b);

• 逻辑与轮询相同,但等待通知是由操作系统完成, cpu 消耗 –

还有问题: 先等 a 再等 b ,但各种 io 事件,谁知道 b 会不会先完成.

解决这个问题还得靠操作系统, wait_until_get_ready 一次只能等待 1 个事件,那能不能一次等待多个?

fn add_to_wait_list(operations: Vec<Operation>) // 加入等待列表
fn wait_until_some_of_them_get_ready() ->Vec<Operation> // 阻塞等待多个事件

因此上面代码可以继续修改:

let op_a = read_async("qq.com");
let op_b = read_async("jd.com");
add_to_wait_list([op_a, op_b]);
while true {
  let list = wait_until_some_of_them_get_ready();
  if list.is_empty() {
    break;
  }
  for op in list {
    if op.equal(op_a) {
      write_to("qq.html", op.get_content());
    } else if op.equal(op_b) {
      write_to("jd.html", op.get_content());
    }
  }
}

• 这次是那个 io 事件先回来,就处理那个了.

还有什么问题吗? –> 事件处理时候还得 一堆 if-elsse,本来 事件处理函数就是和事件绑定的,这么多 if-else 干嘛,几个还好,事件一多成了面条了. –> 回调函数

还是得靠操作系统 , read_async_v1 内绑定事件处理函数 -> read_async_v2 干脆 callback 也一并注册了.

function read_async_v1(targetURL: String, callback: Function) {
  let operation = read_async("qq.com");
  operation.callback = callback;
  return operation;
}

function read_async_v2(target, callback) {
  let operation = read_async(target);
  operation.callback = callback;
  add_to_wait_list([operation]);
}

read_async_v2("qq.com", function(data) {
  send_to("[email protected]", data);
});
read_async_v2("jd.com", function(data) {
  write_to("jd.html", data);
});
while true {
  let list = wait_until_some_of_them_get_ready();
  if list.is_empty() {
    break;
  }
  for op in list {
    op.callback(op.get_content());
  }
}

• 事件处理时候直接回调,多清爽.

事件处理的等待 可以 麻烦操作系统 or 框架

read_async_v2("qq.com", function(data) {
  send_to("[email protected]", data);
});
read_async_v2("jd.com", function(data) {
  write_to("jd.html", data);
});
// 事件等待循环交给操作系统

到这里就是一个基本的异步 IO 思路了,可以对应到很多的异步框架 or 接口,例如 epoll(Linux) 各种异步框架,不过 Linux 下 C 的实现更加繁琐难懂.

回调地狱

到这里就结束了? 怎么可能,欢迎来到 回调地狱!

任务与任务之间会有 前后关系,先处理谁 再处理谁,因此每个事件都绑定回调情况下…

login(user => {
    getStatus(status => {
        getOrder(order => {
            getPayment(payment => {
                getRecommendAdvertisements(ads => {
                    setTimeout(() => {
                        alert(ads)
                    }, 1000)
                })
            })
        })
    })
})

• 初次碰到是在 android,一个按钮按下时候 那个酸爽.

要是写成下面这样就好了….

login(username, password)
    .then(user => getStatus(user.id))
    .then(status => getOrder(status.id))
    .then(order => getPayment(order.id))
    .then(payment => getRecommendAdvertisements(payment.total_amount))
    .then(ads => {/*...*/});

• rxjava kolin的协程 or flow 都是这样的链式调用;
• 可惜早年接触时候,理解尚浅.

read_async_v2 实际调用时候,已经执行完了. 为了支持 .then() 这样的调用

function read_async_v3(target) {
  let operation = read_async(target);
  add_to_wait_list([operation]);
  return {
    then: function(callback) {
        operation.callback = callback;
    },
  }
}
// 我们可以这样
read_async_v3("qq.com").then(logic)

• 返回一个高阶函数,但这样依旧只能绑定一个回调,而不能链式调用. -> 那就弄一个 callback list

function read_async_v4(target) {
  let operation = read_async(target);
  add_to_wait_list([operation]);
  let chainableObject = {
    callbacks: [],
    then: function(callback) {
      this.callbacks.push(callback);
      return this;
    },
    run: function(data) {
      let nextData = data;
      for cb in this.callbacks {
        nextData = cb(nextData);
      }
    }
  };
  operation.callback = chainableObject.run;
  return chainableObject;
}

• 返回一个 chainableObject 对象
• 对象本身保存有 callback list. 每次 then 都是返回自身,因此能一直 then().then()…
• 最后 callback = chainableObject.run,回调时候每次调用的都是 那一次 then 绑定好的 run 方法,就能依次层层调用了.
• 感觉有些类似: 回调地狱是同级的关系 -> 抽象层次再高一级, 层层递归就能写成 list 了.

终于可以这样了 read_async_v4("qq.com").then(logic1).then(logic2).then(/*...*/),但是…

read_async_v4("qq.com")
    .then(data => console.log("qq.com: ${data}"))
    .then((_) => read_async_v4("jd.com"))
    .then(data => console.log("jd.com: {$data}"))

^e6a61a

看起来挺正常的啊,但实际上 3 个 then附加的函数 都添加到了 "qq.com" 的 chainableObject 的 callbacks 中.执行到第二个 callback 时候,是真的执行了 read_async_v4("jd.com") 函数,异步调用返回了 chainableObject,第三步入参就成了 chainableObject 了.
相当于整个链条的 callback 不能有异步操作..这算哪门子的异步..

问题就在于 nextData = cb(nextData); 时候遇到 callback 包含异步操作怎么办.

• 也好办,支持 异步操作 本身需要一个 ChainableObject 包装,然后调用其 run 才能正常等待.
• 这一步其实已经拿到了 ChainableObject,需要改动的就是将剩下的 callbacks 转移到这个 ChainableObject 上,这样剩余的 callbacks 会在新的 ChainableObject 上等待其完成.

function read_async_v5(target) {
  let operation = read_async(target);
  add_to_wait_list([operation]);
  let chainableObject = {
    callbacks: [],
    then: function(callback) {
      this.callbacks.push(callback);
      return this;
    },
    run: function(data) {
      let nextData = data;
       let self = this;
      while self.callbacks.length > 0 {
          // 每次从队首弹出一个回调函数
          let cb = self.callbacks.pop_front();
          nextData = cb(nextData);
          // 如果回调返回了一个ChainableObject，那么就把剩下的callback绑定到它上面
          // 然后就可以终止执行了
          if isChainableType(nextData) {
              nextData.callbacks = self.callbacks;
              return;
          }
      }
    }
  };
  operation.callback = chainableObject.run;
  return chainableObject;
}

• 高阶函数/对象 的嵌套递归就是这样折腾.
• 复杂度不会消失,只是转移.

这样 [[#^e6a61a]] 才能正常调用了.

对于 js 而言,这里的 chainableObject 就是 Promise

• 对 js 不熟,原文如此.

到了这里就能以回调的形式写出 链式调用了.但还远远不是终点呢还有 async/await,尽管将逻辑以 async/await 重新组织并不容易.