compose

题目描述

实现一个 compose 函数，用于函数组合：

• 将一系列函数像管道一样连接起来
• 支持同步和异步函数的混合组合
• 数据从第一个函数流向最后一个函数
• 实现函数式编程中的核心概念

本题考查 函数式编程、异步编程 和 高阶函数 的综合应用。

核心知识点

1. 函数组合的核心概念

// 函数组合的基本思想
const add = (x) => x + 1;
const multiply = (x) => x * 2;
const square = (x) => x * x;

// 传统写法
const result = square(multiply(add(5))); // ((5+1)*2)² = 144

// compose 写法
const composed = compose([add, multiply, square]);
const result = composed(5); // 144

2. 执行顺序（从左到右）

// compose([f, g, h]) 执行顺序：
// input → f(input) → g(f(input)) → h(g(f(input))) → output

// 这与数学中的函数组合 h∘g∘f 相反
// 但更符合直觉的数据流动方向

3. 同步与异步兼容

// 同步函数
const syncFn = x => x * 2

// 异步函数
const asyncFn = x => Promise.resolve(x + 1)

// 混合使用
const mixed = compose([syncFn, asyncFn, syncFn])
// 自动处理同步/异步的差异

4. Promise 链式处理

// 核心原理：将所有函数返回值用 Promise.resolve 包装
// 确保 .then() 方法始终可用
return fns.reduce((promise, fn) => promise.then(fn), Promise.resolve(input));

代码实现

/**
 * 组合多个函数，返回一个新的函数，按从右到左的顺序依次执行这些函数，并支持异步操作。
 *
 * @param {Function[]} fns - 要组合的函数数组。每个函数接收上一个函数的返回值作为参数，可以返回 Promise 或普通值。
 * @returns {Function} 返回一个新的函数，接收任意参数，依次执行组合的函数，返回最终的 Promise。
 */
export default function compose(fns) {
  const init = fns.shift()

  return function (...args) {
    return fns.reduce(
      async (acc, cur) => {
        const result = await acc
        return cur.call(this, result)
      },
      Promise.resolve(init.apply(null, args)),
    )
  }
}

关键技术点

1. Promise.resolve 的妙用

// 统一处理同步和异步返回值
const result = fn(value);
return Promise.resolve(result); // 同步值包装成 Promise

// 这样后续都可以使用 .then() 方法

2. reduce 实现链式调用

// 核心模式：accumulator 是 Promise，currentValue 是函数
return fns.reduce(
  (promise, fn) => promise.then(fn),
  Promise.resolve(initialValue),
);

3. 首个函数的特殊处理

// 第一个函数可以接收多个参数
const [firstFn, ...restFns] = fns
const initialResult = firstFn(...args) // 多参数

// 后续函数只接收一个参数（上一个函数的返回值）
restFns.forEach(fn => (result = result.then(fn)))

4. 错误处理和传播

// Promise 链自动传播错误
promise.then(fn).catch((error) => {
  // 任何环节的错误都会被捕获
  console.error('Function composition error:', error)
  throw error // 继续传播
})

扩展思考

1. 支持分支的组合

function composeBranch(condition, trueBranch, falseBranch) {
  return function (input) {
    return Promise.resolve(condition(input)).then((shouldTakeTrueBranch) => {
      const branch = shouldTakeTrueBranch ? trueBranch : falseBranch
      return compose(branch)(input)
    })
  }
}

// 使用示例
const isEven = x => x % 2 === 0
const evenBranch = [x => x / 2, x => `Even result: ${x}`]
const oddBranch = [x => x * 3 + 1, x => `Odd result: ${x}`]

const conditionalPipeline = composeBranch(isEven, evenBranch, oddBranch)

2. 支持循环的组合

function composeLoop(fns, condition, maxIterations = 100) {
  return function (input) {
    const result = Promise.resolve(input)
    let iterations = 0

    const iterate = (value) => {
      if (iterations >= maxIterations || !condition(value)) {
        return value
      }

      iterations++
      return compose(fns)(value).then(iterate)
    }

    return result.then(iterate)
  }
}

// 使用示例：计算平方根（牛顿法）
const improve = x => y => (y + x / y) / 2
const goodEnough = x => y => Math.abs(y * y - x) < 0.001

function sqrt(x) {
  return composeLoop([improve(x)], goodEnough(x))(1.0)
}

3. 支持记忆化的组合

function composeMemoized(fns, keyGenerator = JSON.stringify) {
  const cache = new Map()
  const composedFn = compose(fns)

  return function (...args) {
    const key = keyGenerator(args)

    if (cache.has(key)) {
      return Promise.resolve(cache.get(key))
    }

    return composedFn(...args).then((result) => {
      cache.set(key, result)
      return result
    })
  }
}

4. 支持并发控制的组合

function composeWithConcurrency(fns, concurrency = 2) {
  return function (...args) {
    const queue = [...fns]
    const results = []
    const running = []

    const executeNext = () => {
      if (queue.length === 0) {
        return Promise.all(running)
      }

      const fn = queue.shift()
      const promise = Promise.resolve(fn(...args))
      running.push(promise)

      if (running.length >= concurrency) {
        return Promise.race(running).then(() => {
          const index = running.findIndex(p => p === promise)
          running.splice(index, 1)
          return executeNext()
        })
      }

      return executeNext()
    }

    return executeNext()
  }
}

5. 支持流式处理的组合

function composeStream(fns) {
  return function (inputStream) {
    return fns.reduce((stream, fn) => {
      return stream.pipe(
        new Transform({
          objectMode: true,
          transform(chunk, encoding, callback) {
            try {
              const result = fn(chunk)

              if (result && typeof result.then === 'function') {
                result
                  .then(value => callback(null, value))
                  .catch(error => callback(error))
              }
              else {
                callback(null, result)
              }
            }
            catch (error) {
              callback(error)
            }
          },
        }),
      )
    }, inputStream)
  }
}