最大并行任务池

[Lucifier129]

编程面试题之最大并行任务池：

• 1）给定一个 taskList 和 poolSize；
• 2）task 是异步的，返回 promise；
• 3）task 不会失败。要求：
  • 1）同时不能有超过 poolSize 的 task 在执行；
  • 2）所有 taskList 里的 task 最终都执行完毕。

难度升级：taskList 不是给定的数组，而是不定的任务流，可以用 async iterator 来表达。实例场景之一是，浏览器流式解析 html ，异步的发现 css, js, image, ajax 等资源请求，同时控制最大并行请求的数量。

[Lucifier129]

const createDeferred = () => {
  let resolve, reject
  let promise = new Promise((a, b) => {
    resolve = a
    reject = b
  })
  return { promise, resolve, reject }
}

const createAsyncList = (...args) => {
  let index = 0
  let valueList = args
  let deferredList = []
  let consume = () => {
    let currentIndex = index++
    let item = valueList[currentIndex]
    if (item) return item.value

    let deferred = deferredList[currentIndex]
    if (!deferred) {
      deferred = deferredList[currentIndex] = createDeferred()
    }

    return deferred.promise
  }

  let count = 0
  let next = value => {
    if (!value) throw new Error('value should be a function')
    let index = count++
    let deferred = deferredList[index]
    valueList[index] = { value }
    deferred && deferred.resolve(value)
  }

  return { consume, next }
}

const sequence = async getNextTask => {
  let task = await getNextTask()
  if (!task) return
  await task()
  await sequence(getNextTask)
}

const parallel = async (getNextTask, size) => {
  let taskList = await Promise.all(Array.from({ length: size }, getNextTask))
  return Promise.all(taskList.filter(Boolean).map(task => task()))
}

const buffer = size => {
  let { consume, next } = createAsyncList()
  parallel(() => () => sequence(consume), size)
  return next
}

const sleep = ms => new Promise(resolve => setTimeout(resolve, ms))

const test = async () => {
  let next = buffer(3)
  let count = 0
  while (true) {
    let current = count++

    await sleep(Math.random() * 300)

    next(async () => {
      await sleep(100)
      console.log(current)
    })
  }
}

test()

[Lucifier129]

async function sequence(taskList) {
  for await (let task of taskList) {
    if (!task) return
    await task()
  }
}

async function buffer(taskList, size) {
  let pool = []

  while (true) {
    if (size <= 0) return
    pool.push(sequence(taskList))
    size -= 1
  }

  await Promise.all(pool)
}

const sleep = ms => new Promise(resolve => setTimeout(resolve, ms))

const createTaskList = async function*() {
  let count = 0
  while (true) {
    let current = count++
    let task = async () => {
      await sleep(100)
      console.log(current)
    }

    await sleep(Math.random() * 300)
    yield task
  }
}

buffer(createTaskList(), 3)

[Lucifier129]

支持使用 for-await of 依次获取 task 的返回值，而 taskList 里的计算，则是以 size 为单位的并行执行。

async function sequence(taskList) {
  for await (let task of taskList) {
    if (!task) return
    await task()
  }
}

async function buffer(taskList, size) {
  let pool = []

  while (true) {
    if (size <= 0) return
    pool.push(sequence(taskList))
    size -= 1
  }

  await Promise.all(pool)
}

const createDeferred = () => {
  let resolve, reject
  let promise = new Promise((a, b) => {
    resolve = a
    reject = b
  })
  return { promise, resolve, reject }
}

const map = async function*(list, mapper) {
  let count = 0
  for await (let item of list) {
    yield mapper(item, count++)
  }
}

const consumeBuffer = async function*(taskList, size) {
  let deferredList = []
  let getDeferred = index => {
    let deferred = deferredList[index]
    if (deferred) return deferred
    deferredList[index] = createDeferred()
    return getDeferred(index)
  }
  let newTaskList = map(taskList, (task, index) => {
    if (!task) {
      getDeferred(index).resolve()
      return
    }
    return async () => {
      let value = await task()
      getDeferred(index).resolve(value)
      return value
    }
  })

  buffer(newTaskList, size)

  let count = 0
  while (true) {
    yield await getDeferred(count++).promise
  }
}

const sleep = ms => new Promise(resolve => setTimeout(resolve, ms))

const createTaskList = async function*() {
  let count = 0
  while (true) {
    let value = count++
    let task = async () => {
      await sleep(1000)
      return value
    }

    await sleep(Math.random() * 30)
    yield task
  }
}

const test = async function() {
  for await (let value of consumeBuffer(createTaskList(), 3)) {
    console.log(value)
  }
}

test()

[Lucifier129]

pushable producer & pullable consumer

const pipe = (...args) => args.reduce((a, f) => f(a))
const sleep = ms => new Promise(resolve => setTimeout(resolve, ms))

const create = producer => sink => {
  let count = 0
  let valueList = []
  let consumerList = []

  producer(value => {
    let index = valueList.length
    valueList.push(value)

    if (index < consumerList.length) {
      let consumer = consumerList[index]
      consumerList[index] = null
      consumer(value)
    }
  })

  sink(consumer => {
    let index = consumerList.length
    consumerList.push(consumer)

    if (index < valueList.length) {
      consumerList[index] = null
      consumer(valueList[index])
    }
  })
}

let parallel = size => source => sink =>
  source(next => {
    while (size > 0) {
      sink(next)
      size -= 1
    }
  })

const foreach = f => source =>
  source(next => {
    let consumer = async value => {
      await f(value)
      next(consumer)
    }
    next(consumer)
  })

const task$ = create(async next => {
  let count = 0
  while (true) {
    let value = count++
    let task = async () => {
      await sleep(1000)
      return value
    }

    await sleep(Math.random() * 30)
    next(task)
  }
})

pipe(
  task$,
  parallel(3),
  foreach(async task => {
    let result = await task()
    console.log(result)
  })
)

[colinNaive]

watch一下 要不有新题 都接不到通知 哈哈

[Lucifier129]

基于 ES2015 Classes 的实现

const createDeferred = () => {
  let resolve, reject
  let promise = new Promise((a, b) => {
    resolve = a
    reject = b
  })
  return { promise, resolve, reject }
}

class DeferredList {
  constructor() {
    this.count = 0
    this.list = []
  }
  get(index) {
    let item = this.list[index]
    if (item) return item
    this.list.push(createDeferred())
    return this.get(index)
  }
  resolve(index, value) {
    return this.get(index).resolve(value)
  }
  reject(index, reason) {
    return this.get(index).reject(reason)
  }
}

class AsyncTaskList {
  constructor() {
    this.count = 0
    this.index = 0
    this.deferredList = new DeferredList()
  }
  async next() {
    let { deferredList, count } = this
    this.count += 1
    return deferredList.get(count).promise
  }
  add(task) {
    let { deferredList, index } = this
    this.index += 1
    deferredList.resolve(index, task)
  }
}

class SequenceTask {
  constructor() {
    this.taskList = new AsyncTaskList()
  }
  async consume() {
    let { taskList } = this
    let task = await taskList.next()

    if (!task) return
    await task()
    await this.consume()
  }
  add(task) {
    this.taskList.add(task)
  }
}

class ConcurrencyTask {
  constructor(limit = 1) {
    this.limit = limit
    this.sequence = new SequenceTask()
  }
  async start() {
    let { limit, sequence } = this
    let asyncList = []

    for (let i = 0; i < limit; i++) {
      asyncList.push(sequence.consume())
    }

    await Promise.all(asyncList)
  }
  add(task) {
    this.sequence.add(task)
  }
}

const sleep = ms => new Promise(resolve => setTimeout(resolve, ms))

const createTaskList = async function*() {
  let count = 0
  while (true) {
    let value = count++
    let task = async () => {
      await sleep(1000)
      console.log(value)
    }

    await sleep(Math.random() * 30)
    yield task
  }
}

const test = async function() {
  let concurrency = new ConcurrencyTask(3)

  concurrency.start()
  for await (let task of createTaskList()) {
    concurrency.add(task)
  }
}

test()

[Lucifier129]

全用 const 声明变量和函数，处理给定的 taskList

const foldAsync = async (list = [], f, acc) => {
  if (list.length === 0) return acc
  const [value, ...remain] = list
  return foldAsync(remain, f, await f(acc, value))
}

const indexedAsyncList = list => list.map(async (promise, index) => {
  await promise
  return index
})

const parallel = async (taskList, size = 1) => {
  const [totalList, pendingList] = await foldAsync(taskList, async ([totalList, pendingList], task) => {
    if (pendingList.length < size) {
      const promise = task()
      return [totalList.concat(promise), pendingList.concat(promise)]
    }

    const indexedList = indexedAsyncList(pendingList)
    const fastestIndex = await Promise.race(indexedList)
    const remainList = pendingList.filter((_, i) => i !== fastestIndex)
    const promise = task()

    return [totalList.concat(promise), remainList.concat(promise)]
  }, [[], []])

  await Promise.all(pendingList)
  return Promise.all(totalList)
}

const main = async () => {
  console.time('main');
  const p1 = () => new Promise((res, rej) => setTimeout(() => res('a'), 1000));
  const p2 = () => new Promise((res, rej) => setTimeout(() => res('b'), 3000));
  const p3 = () => new Promise((res, rej) => setTimeout(() => res('c'), 2000));
  const p4 = () => new Promise((res, rej) => setTimeout(() => res('d'), 5000));
  const p5 = () => new Promise((res, rej) => setTimeout(() => res('e'), 4000));
  const k1 = [p1, p2, p3, p4, p5];
  // 最大并发数为3
  const results = await parallel(k1, 3);
  console.log(results);
  console.timeEnd('main');
};
main();  // ["a", "b", "c", "d", "e"] 6008.335693359375ms

[Elity]

大佬，@Lucifier129 下面这种情况难度算是降低了还是提高了？任务来自于外界添加，可能是同步，也可能是异步。

---

完成Scheduler代码，实现一个带并发限制的异步调度器，保证同时最多运行Scheduler.max个任务

class Scheduler {
  max = 2
  add(task) {
  }
}

const timeout = time => new Promise(r => setTimeout(r, time));

const scheduler = new Scheduler();

const addTask = (time, log) => scheduler.add(() => timeout(time)).then(() => console.log(log));

addTask(1000, 1);
addTask(500, 2);
addTask(300, 3);
addTask(400, 4);

// 依次输出 2 3 1 4

[Lucifier129]

@Elity 跟题目难度一样啊，task 是动态添加的，同时最多只有 n 个在执行。

[zhanba]

为啥限定task不会失败？