别名:: transducer：强大的函数组合模式 笔记本:: transducer: 一种强大的函数组合模式
地图和过滤器
map 的作用是对集合中的每个元素应用一个转换函数。
<code class="javascript">const list = [1, 2, 3, 4, 5]; list.map(x => x + 1); // [2, 3, 4, 5, 6]</code>
为了更清晰地展示 map 的实现，我们使用一个 for 循环：
<code class="javascript">function map(f, xs) {
const ret = [];
for (let i = 0; i < xs.length; i++) {
ret.push(f(xs[i]));
}
return ret;
}
map(x => x + 1, [1, 2, 3, 4, 5]);
// [2, 3, 4, 5, 6]</code>
这段代码明确说明了 map 的实现依赖于数组的特性：顺序执行，立即求值。
接下来看看 filter：
<code class="javascript">function filter(f, xs) {
const ret = [];
for (let i = 0; i < xs.length; i++) {
if (f(xs[i])) {
ret.push(xs[i]);
}
}
return ret;
}
const range = n => [...Array(n).keys()];
filter(x => x % 2 === 1, range(10));
// [1, 3, 5, 7, 9]</code>
同样，filter 也依赖于数组类型。
如何让 map 等函数支持不同数据类型，例如集合、Map 和自定义类型？一种通用的方法是依赖集合的接口（协议）。不同语言的实现方式不同。JavaScript 原生支持较弱，但可以通过 Symbol.iterator 进行迭代和使用 Object.constructor 获取构造函数来实现。
为了更灵活地处理不同数据类型，我们可以模仿 RamdaJS 库，使用自定义的 @@transducer/step 函数：
<code class="javascript">function map(f, xs) {
const ret = new xs.constructor(); // 1. 构造函数
for (const x of xs) { // 2. 迭代
ret['@@transducer/step'](f(x)); // 3. 收集
}
return ret;
}
Array.prototype['@@transducer/step'] = Array.prototype.push;
// [function: push]
map(x => x + 1, [1, 2, 3, 4, 5]);
// [2, 3, 4, 5, 6]
Set.prototype['@@transducer/step'] = Set.prototype.add;
// [function: add]
map(x => x + 1, new Set([1, 2, 3, 4, 5]));
// Set(5) { 2, 3, 4, 5, 6 }</code>
通过这种方式，我们实现了更通用的 map 函数。关键在于将构造、迭代和收集操作委托给具体的集合类型，因为只有集合本身才知道如何执行这些操作。
类似地，我们可以实现一个更通用的 filter 函数：
<code class="javascript">function filter(f, xs) {
const ret = new xs.constructor();
for (const x of xs) {
if (f(x)) {
ret['@@transducer/step'](x);
}
}
return ret;
}
filter(x => x % 2 === 1, range(10));
// [1, 3, 5, 7, 9]
filter(x => x > 3, new Set(range(10)));
// Set(6) { 4, 5, 6, 7, 8, 9 }</code>
组合
将 map 和 filter 组合使用时，会出现一些问题：
<code class="javascript">range(10) .map(x => x + 1) .filter(x => x % 2 === 1) .slice(0, 3); // [1, 3, 5]</code>
尽管只使用了前三个元素，但整个集合都被遍历了，并且生成了多个中间集合对象。为了解决这个问题，我们使用 compose 函数：
<code class="javascript">function compose(...fns) {
return fns.reduceRight((acc, fn) => x => fn(acc(x)), x => x);
}</code>
为了支持组合，我们将 map 和 filter 函数进行柯里化：
<code class="javascript">function curry(f) {
return (...args) => data => f(...args, data);
}
const rmap = curry(map);
const rfilter = curry(filter);</code>
我们还需要一个 take 函数：
<code class="javascript">function take(n, xs) {
const ret = new xs.constructor();
for (const x of xs) {
if (n-- > 0) {
ret['@@transducer/step'](x);
}
}
return ret;
}
take(3, range(10));
// [0, 1, 2]
take(4, new Set(range(10)));
// Set(4) { 0, 1, 2, 3 }</code>
现在我们可以组合这些函数了：
<code class="javascript">const rtake = curry(take); const takefirst3odd = compose( rtake(3), rfilter(x => x % 2 === 1), rmap(x => x + 1) ); takefirst3odd(range(10)); // [1, 3, 5]</code>
虽然代码清晰简洁，但运行效率仍然不高。
函数类型
Transducer
柯里化后的 map 函数的类型如下：
<code class="typescript">const map = f => xs => ...</code>
它返回一个单参数函数，这个函数就是一个 transformer：
<code class="typescript">type transformer = (xs: T) => R;</code>
transformer 方便函数组合。它的输入是数据，输出是处理后的数据。
<code class="typescript">data -> map(...) -> filter(...) -> reduce(...) -> result</code>
我们可以使用 pipe 函数来组合 transformer：
<code class="javascript">function pipe(...fns) {
return x => fns.reduce((ac, f) => f(ac), x);
}
const reduce = (f, init) => xs => xs.reduce(f, init);
const f = pipe(
rmap(x => x + 1),
rfilter(x => x % 2 === 1),
rtake(5),
reduce((a, b) => a + b, 0)
);
f(range(100));
// 25</code>
transformer 是单参数函数，方便组合。
Reducer
reducer 是一个双参数函数，可以表达更复杂的逻辑：
<code class="typescript">type reducer = (ac: R, x: T) => R;</code>
加法
<code class="javascript">// add is an reducer const add = (a, b) => a + b; const sum = xs => xs.reduce(add, 0); sum(range(11)); // 55</code>
Map
<code class="javascript">function concat(list, x) {
list.push(x);
return list;
}
const map = f => xs => xs.reduce((ac, x) => concat(ac, f(x)), []);
map(x => x * 2)(range(10));
// [0, 2, 4, 6, 8, 10, 12, 14, 16, 18]</code>
Filter
<code class="javascript">const filter = f => xs => xs.reduce((ac, x) => f(x) ? concat(ac, x) : ac, []); filter(x => x > 3 && x < 7, range(10)); // [4, 5, 6]</code>
Take
实现 take 需要 reduce 函数具有类似 break 的功能：
<code class="javascript">function reduced(x) {
return x && x['@@transducer/reduced'] ? x : { '@@transducer/reduced': true, '@@transducer/value': x };
}
function reduce(f, init) {
return xs => {
let ac = init;
for (const x of xs) {
const r = f(ac, x);
if (r && r['@@transducer/reduced']) {
return r['@@transducer/value'];
}
ac = r;
}
return ac;
};
}
function take(n) {
return xs => {
let i = 0;
return reduce((ac, x) => {
if (i === n) {
return reduced(ac);
}
i++;
return concat(ac, x);
}, [])(xs);
};
}
take(4)(range(10));
// [0, 1, 2, 3]</code>
Transducer
让我们重新审视 map 函数：
<code class="javascript">function map(f, xs) {
const ret = [];
for (let i = 0; i < xs.length; i++) {
ret.push(f(xs[i]));
}
return ret;
}</code>
我们需要将依赖数组的逻辑抽象成一个 reducer：
<code class="javascript">function rmap(f) {
return reducer => {
return (ac, x) => {
return reducer(ac, f(x));
};
};
}</code>
构造、迭代和收集操作都消失了。rmap 只包含其核心逻辑。
类似地，我们可以实现 rfilter：
<code class="javascript">function rfilter(f) {
return reducer => (ac, x) => {
return f(x) ? reducer(ac, x) : ac;
};
}</code>
注意 rfilter 和 rmap 的返回类型：
<code class="typescript">reducer => (acc, x) => ...</code>
它是一个 transducer，参数和返回值都是 reducer。transducer 是可组合的。
应用 Transducer
如何使用 transducer？
<code class="javascript">const compose = (...fns) => fns.reduceRight((acc, fn) => x => fn(acc(x)), x => x);
const tf = compose(
rmap(x => x + 1),
rfilter(x => x % 2 === 1),
rtake(5)
);
const collect = (ac, x) => {
ac.push(x);
return ac;
};
const reducer = tf(collect);
reduce(reducer, [])(range(100));
// [1, 3, 5, 7, 9]</code>
我们将逻辑封装成一个函数：
<code class="javascript">function into(init, tf) {
const reducer = tf(collect);
return reduce(reducer, init);
}
into([], compose(
rmap(x => x + 1),
rfilter(x => x % 2 === 1),
rtake(8)
))(range(100));
// [1, 3, 5, 7, 9, 11, 13, 15]</code>
迭代是按需的。
异步数据流
考虑一个异步的斐波那契数列生成器：
<code class="javascript">function sleep(n) {
return new Promise(r => setTimeout(r, n));
}
async function* fibs() {
let [a, b] = [0, 1];
while (true) {
await sleep(10);
yield a;
;[a, b] = [b, a + b];
}
}
const s = fibs();
async function start() {
let i = 0;
for await (const item of s) {
console.log(item);
i++;
if (i > 10) {
break;
}
}
}
start();</code>
我们需要修改 into 函数以支持异步迭代器：
<code class="javascript">const collect = (ac, x) => {
ac.push(x);
return ac;
};
const reduce = (reducer, init) => {
return async iter => {
let ac = init;
for await (const item of iter) {
if (ac && ac['@@transducer/reduced']) {
return ac['@@transducer/value'];
}
ac = reducer(ac, item);
}
return ac;
};
};
function sinto(init, tf) {
const reducer = tf(collect);
return reduce(reducer, init);
}
const task = sinto([], compose(
rmap(x => x + 1),
rfilter(x => x % 2 === 1),
rtake(8)
));
task(fibs()).then(res => {
console.log(res);
});</code>
相同的逻辑适用于不同的数据结构。
执行顺序
基于柯里化的 compose 和基于 reducer 的 compose 的参数顺序不同。
柯里化版本
函数执行是右关联的。
Transducer 版本
参考
Transducer - Clojure 参考
