Реализация и расширение асинхронной очереди JavaScript

представлять

Для любого языка важны очереди, сериализация io, параллелизм запросов и т.д. В JavaScript асинхронное программирование очень важно из-за одного потока. Вдохновленный вчерашним вопросом на собеседовании, когда я реализовывал асинхронную очередь на JS, я позаимствовал идею экспресс-мидлвара и распространил ее на совместную реализацию и генератор, а также на asyncToGenerator.

ps: В этой статье нет картинок, а кода много. Этот код варианта использованияздесь, можешь клонировать и попробовать

Асинхронная очередь

Во многих интервью задают вопрос, как заставить асинхронные функции выполняться последовательно. Существует множество методов, таких как обратный вызов, обещание, наблюдатель, генератор, асинхронный/ожидающий, все из которых имеют дело с асинхронным программированием в JS и могут удовлетворить это последовательное требование. Но беда в том, что с ней очень хлопотно разбираться, приходится вручную вызывать следующую задачу в предыдущей задаче. Как и обещания, вот так:

a.then(() => b.then(() => c.then(...)))

Проблема вложенности кода немного серьезна. Поэтому, если есть очередь, просто добавьте в нее асинхронные задачи и позвольте очереди запускаться при ее выполнении. Давайте сначала сформулируем API.У нас есть очередь, и очередь поддерживается внутри.Добавляем асинхронные задачи через queue.add, и queue.run выполняет очередь.Вы можете подумать об этом в первую очередь.

Обратитесь к предыдущему экспрессупромежуточное ПОРеализация асинхронной задачи async-fun проходит в методе next, и только при вызове next очередь продолжит опускаться. Тогда это следующее очень важно, оно будет управлять очередью, чтобы переместиться на один бит назад и выполнить следующее асинхронное развлечение. Нам нужна очередь для асинхронного веселья и курсор для управления порядком.

Вот моя простая реализация:

const queue = () => {
  const list = []; // 队列
  let index = 0;  // 游标

  // next 方法
  const next = () => {
    if (index >= list.length - 1) return;    

    // 游标 + 1
    const cur = list[++index];
    cur(next);
  }

  // 添加任务
  const add = (...fn) => {
    list.push(...fn);
  }

  // 执行
  const run = (...args) => {
    const cur = list[index];
    typeof cur === 'function' && cur(next);
  }

  // 返回一个对象
  return {
    add,
    run,
  }
}

// 生成异步任务
const async = (x) => {
  return (next) => {// 传入 next 函数
    setTimeout(() => {
      console.log(x);
      next();  // 异步任务完成调用
    }, 1000);
  }
}

const q = queue();
const funs = '123456'.split('').map(x => async(x));
q.add(...funs);
q.run();// 1, 2, 3, 4, 5, 6 隔一秒一个。

Я не конструировал здесь класс, а разобрался с ним через характеристики замыканий. Метод очереди возвращает объект, содержащий add и run, Add — добавить асинхронный метод в очередь, run — запустить выполнение. Внутри очереди мы определяем несколько переменных, list используется для сохранения очереди, index — это курсор, который указывает, в какую функцию сейчас ушла очередь, кроме того, наиболее важным является метод next, который управляет перемещением курсора назад.

После выполнения функции запуска очередь начинает выполняться. В начале выполнения первой асинхронной функции в очереди мы передаем ей следующую функцию, а затем асинхронная функция решает, когда выполнить следующую, то есть приступить к выполнению следующей задачи. Мы не знаем, когда асинхронная задача будет завершена, и мы можем только сообщить очереди, что задача завершена, достигнув определенного консенсуса. Это следующая функция, переданная задаче. На самом деле функция, возвращаемая async, имеет имя Thunk, которое мы кратко представим позже.

Thunk

Преобразователь на самом деле решить "позвонить по имени". То есть я передаю выражение в функцию А как параметр x+1, но я не уверен, когда этот x+1 будет использован, и будет ли он использован, если он передан и выполнен, то эта оценка не необходимо . Итак, существует временная функция Thunk для сохранения этого выражения, передачи его в функцию A и вызова при необходимости.

const thunk = () => {
  return x + 1;
};

const A = thunk => {
  return thunk() * 2;
}

Ну... на самом деле это функция обратного вызова...

Пауза

На самом деле, пока задача не продолжает вызывать следующую, очередь не будет продолжать опускаться. Например, добавляем суждение в асинхронную задачу (обычно асинхронный io, отказоустойчивая обработка запросов):

// queue 函数不变，
// async 加限制条件
const async = (x) => {
  return (next) => {
    setTimeout(() => {
      if(x > 3) {
        console.log(x);
        q.run();  //重试
        return;
      }
      console.log(x);
      next();
    }, 1000);
  }
}

const q = queue();
const funs = '123456'.split('').map(x => async(x));
q.add(...funs);
q.run();
//打印结果： 1, 2, 3, 4, 4，4， 4，4 一直是 4

Когда выполняется четвертая задача, когда x равно 4, если вы не продолжите, вы можете вернуться напрямую, не вызывая next. Также может быть ошибка, нужно попробовать еще раз, затем снова вызвать q.run, потому что курсор сохраняет индекс текущей асинхронной задачи.

Кроме того, есть еще один способ добавить метод остановки. Хотя описанный выше метод выглядит нормально, преимущество остановки заключается в том, что вы можете активно останавливать очередь вместо добавления ограничений на асинхронные задачи. Конечно, если есть пауза, она продолжится.Есть два пути.Один – это повтор, то есть повторное выполнение того, что было приостановлено в прошлый раз, и другой – goOn, который переходит к следующему, независимо от того, последний. Над кодом:

const queue = () => {
  const list = [];
  let index = 0;
  let isStop = false;

  const next = () => {
    // 加限制
    if (index >= list.length - 1 || isStop) return;    
    const cur = list[++index];
    cur(next);
  }

  const add = (...fn) => {
    list.push(...fn);
  }

  const run = (...args) => {
    const cur = list[index];
    typeof cur === 'function' && cur(next);
  }

  const stop = () => {
    isStop = true;
  }

  const retry = () => {
    isStop = false;
    run();
  }

  const goOn = () => {
    isStop = false;
    next();
  }

  return {
    add,
    run,
    stop,
    retry,
    goOn,
  }
}

const async = (x) => {
  return (next) => {
    setTimeout(() => {
      console.log(x);
      next();
    }, 1000);
  }
}

const q = queue();
const funs = '123456'.split('').map(x => async(x));
q.add(...funs);
q.run();

setTimeout(() => {
  q.stop();
}, 3000)


setTimeout(() => {
  q.goOn();
}, 5000)

На самом деле перехват все-таки добавляется... Просто из асинхронной функции ее меняют на следующую функцию, а для переключения true/false используется переменная isStop, а переключение приостанавливается. Я добавил два таймера, один на паузу через 3 секунды, другой на продолжение через 5 секунд, (пожалуйста, игнорируйте ошибку таймера), это должно быть, когда очередь достигает трех секунд, то есть выполняется третья задача. , затем через 2 секунды продолжите. Когда результат печатается до 3, он останавливается и через две секунды продолжается до 4, 5 и 6.

Два способа мышления, пожалуйста, подумайте о проблеме в сочетании со сценой.

параллелизм

Все вышеперечисленное выполняется последовательно, если я хочу работать параллельно... Это также очень просто, просто запустите очередь за один раз.

// 为了代码短一些，把 retry，goOn 先去掉了。

const queue = () => {
  const list = [];
  let index = 0;
  let isStop = false;
  let isParallel = false;

  const next = () => {
    if (index >= list.length - 1 || isStop || isParallel) return;    
    const cur = list[++index];
    cur(next);
  }

  const add = (...fn) => {
    list.push(...fn);
  }

  const run = (...args) => {
    const cur = list[index];
    typeof cur === 'function' && cur(next);
  }

  const parallelRun = () => {
    isParallel = true;
    for(const fn of list) {
      fn(next);
    }
  }

  const stop = () => {
    isStop = true;
  }

  return {
    add,
    run,
    stop,
    parallelRun,
  }
}

const async = (x) => {
  return (next) => {
    setTimeout(() => {
      console.log(x);
      next();
    }, 1000);
  }
}

const q = queue();
const funs = '123456'.split('').map(x => async(x));
q.add(...funs);
q.parallelRun();
// 一秒后全部输出 1, 2, 3, 4, 5, 6

Добавил метод parallelRun для распараллеливания, думаю его лучше не ставить в функцию run, абстрактный блок должен быть максимально детализирован. Затем добавляется переменная isParallel, которая по умолчанию false.Учитывая, что может быть вызвана следующая функция, нужно добавить перехват, чтобы не было хаоса.

Выше приведен асинхронный контроллер очереди, очередь, которая реализована с использованием только thunk функций и далее.Можно поменять все коды es6 на es5 для обеспечения совместимости.Конечно, это достаточно просто и не подходит для ответственных сценариев 😆, просто предлагайте идеи .

генератор и ко

Зачем вводить генератор? Во-первых, он также используется для решения асинхронных обратных вызовов. Кроме того, его метод использования заключается в вызове следующей функции, и генератор будет выполняться вниз. По умолчанию это приостановленное состояние. yield эквивалентен q.add выше, добавляя задачи в очередь. Поэтому я также намерен представить вместе, чтобы лучше расширить наше мышление. Дивергентное мышление, суммируйте сходные точки знаний, и вот однажды у вас вдруг появится одно: оказывается, что это так, оказывается, что ххх заимствует у саб ууу, и тогда вы идете изучать ууу--.

Представляем генераторы

Кратко представьте и повторите, потому что некоторые студенты не используют его часто, обязательно будут забывания.

// 一个简单的栗子，介绍它的用法

function* gen(x) {
  const y = yield x + 1;
  console.log(y, 'here'); // 12
  return y;
}

const g = gen(1);
const value = g.next().value; // {value: 2, done: false}

console.log(value); // 2
console.log(g.next(value + 10)); // {value: 12, done: true}

Во-первых, генератор на самом деле является итеративным алгоритмом, определенным внутри тела функции, а затем возвращает объект-итератор. оiterator, вы можете увидеть мою другую статью.
Выполнение gen возвращает объект g, а не результат. g Как и другие итераторы, при вызове метода next гарантируется курсор + 1, и возвращается объект, который содержит значение (результат оператора yield) и выполнено (независимо от того, завершен ли итератор). Кроме того, значение оператора yield, такое как y в приведенном выше коде, является параметром, передаваемым при следующем вызове next, т. е. значение + 10, то есть 12. Такой дизайн выгоден, потому что тогда вы можете , внутри генератора, При задании итерационного алгоритма получается последний результат (или обработанный результат).

Но у генератора есть тот недостаток, что он не будет выполняться автоматически.God TJ написал co для автоматического запуска генератора, то есть автоматического вызова next. Он требует, чтобы функция/инструкция после yield была функцией thunk или объектом promise, потому что только таким образом она будет выполняться последовательно, что совпадает с идеей реализации очереди в начале. Есть две идеи реализации co, одна — thunk, другая — promise, давайте попробуем их все.

Преобразователь реализация

Помните, как была реализована начальная очередь, функция next была определена внутренне, чтобы обеспечить продвижение курсора, а асинхронная функция получала следующий запрос и выполнялась следующим. Здесь то же самое.Нам нужно только определить ту же следующую функцию внутри функции со, чтобы обеспечить продолжение выполнения.Тогда генератор не предоставляет индекс, но предоставляет функцию g.next, поэтому нам нужно передать только асинхронную функцию , Если g.next не подходит, async является оператором после yield, который равен g.value. Но вы не можете пройти g.next напрямую, почему? Поскольку следующую функцию преобразователя необходимо получить через значение возвращаемого значения g.next, если значения нет, следующая функция преобразователя исчезнет... Поэтому нам все еще нужно определить следующую функцию для ее переноса.

Над кодом:

const coThunk = function(gen, ...params) {

  const g = gen(...params);

  const next = (...args) => { // args 用于接收参数
    const ret = g.next(...args);   // args 传给 g.next，即赋值给上一个 yield 的值。
    if(!ret.done) { // 去判断是否完成
      ret.value(next);  // ret.value 就是下一个 thunk 函数
    }
  }

  next(); // 先调用一波
}

// 返回 thunk 函数的 asyncFn
const asyncFn = (x) => {
  return (next) => { // 接收 next
    const data = x + 1;
    setTimeout(() => {
      next && next(data);
    }, 1000)
  }
}

const gen = function* (x) {
  const a = yield asyncFn(x);
  console.log(a);

  const b = yield asyncFn(a);
  console.log(b);

  const c = yield asyncFn(b);
  console.log(c);

  const d = yield asyncFn(c);
  console.log(d);

  console.log('done');
}

coThunk(gen, 1);
// 2, 3, 4, 5, done

Определенная здесь функция gen очень проста, она заключается в передаче параметра 1, а затем каждая asyncFn накапливается асинхронно, то есть сериализуются несколько асинхронных операций, и следующая зависит от возвращаемого значения предыдущей.

выполнение обещания

По сути, идея та же, за исключением того, что вызывается next, и он меняется на внутреннюю часть co. Поскольку оператор после yield является объектом обещания, мы можем получить его внутри co, а затем вg.next().valueОператор then выполняется просто отлично.

// 定义 co
const coPromise = function(gen) {
// 为了执行后的结果可以继续 then
  return new Promise((resolve, reject) => {
    const g = gen();

    const next = (data) => { // 用于传递，只是换个名字
      const ret = g.next(data);
      if(ret.done) { // done 后去执行 resolve，即co().then(resolve)
        resolve(data); // 最好把最后一次的结果给它
        return;
      }
      ret.value.then((data) => { // then 中的第一个参数就是 promise 对象中的 resolve，data 用于接受并传递。
        next(data);  //调用下一次 next
      })
    }

    next();
  })
}

const asyncPromise = (x) => {
  return new Promise((resolve) => {
    setTimeout(() => {
      resolve(x + 1);
    }, 1000)
  })
}

const genP = function* () {
  const data1 = yield asyncPromise(1);
  console.log(data1);

  const data2 = yield asyncPromise(data1);
  console.log(data2);

  const data3 = yield asyncPromise(data2);
  console.log(data3);
}

coPromise(genP).then((data) => {
  setTimeout(() => {
    console.log(data + 1); // 5
  }, 1000)
});
// 一样的 2, 3, 4, 5

На самом деле исходный код co реализуется с помощью этих двух идей, но он больше обрабатывает ошибки и поддерживает вас, чтобы получить массив, объект и реализовать его через promise.all. Кроме того, когда вызывается функция yield thunk, она преобразуется в обещание, с которым нужно работать. Если интересно, можете посмотретьco, я думаю, теперь должно быть ясно.

async/await

Сейчас наиболее часто используемое асинхронное решение в JS, но асинхронность тоже основана на реализации генератора, но она инкапсулирована. Если вы конвертируете async/await в generate/yield, вам нужно всего лишь заменить синтаксис await на yield, затем добавить его в функцию генерации и заменить асинхронное выполнение на coPromise(gennerate) .

const asyncPromise = (x) => {
  return new Promise((resolve) => {
    setTimeout(() => {
      resolve(x + 1);
    }, 1000)
  })
}

async function fn () {
  const data = await asyncPromise(1);
  console.log(data);
}
fn();

// 那转化成 generator 可能就是这样了。 coPromise 就是上面的实现
function* gen() {
  const data = yield asyncPromise(1);
  console.log(data);
}

coPromise(gen);

asyncToGenerator является таким принципом, и на самом деле babel тоже трансформируется таким образом.

наконец

Сначала я реализовал очередь (решение для асинхронной очереди), распространенное в JS, через промежуточную идею экспресса, а затем перешел к реализации простого coThunk и, наконец, заменил thunk на promise. Поскольку асинхронные решения очень важны в JS, при использовании готовых решений, если вы сможете глубоко подумать о принципах реализации, я считаю, что это поможет нам учиться и развиваться.

Добро пожаловать в личный блог звезды:GitHub.com/Сунь Юнцзянь…😜