language-ext Pipes 全解析：在 C 中同时获得 Effects、Streaming 与 Composability

• 后端

导读

Pipes是 language-ext 中用于纯函数式流式编程的核心抽象：它以单个统一的PipeT类型为基础，派生出Producer、Consumer、Pipe、Effect四类可相互连接的流处理组件，让你在常量内存中边流式处理边穿插副作用（Effects），同时保持组件级可组合性。读完本文，你将理解 Pipes 的“三选二”困局解法、yield/awaiting两个原语、|操作符的横向融合原理、Run()的启动方式，并能用仓库内真实示例（文本文件分块流）搭建自己的管线。

这是一个相对复杂的概念，初次接触感到困惑是正常的。仓库中提供了一个完整可运行的示例：Samples/EffectsExamples/Examples/TextFileChunkStreamExample.cs，建议配合本文一起阅读。

一、传统流编程的“三选二”困局

先看文档指出的核心痛点：常规的流编程迫使你在以下三个特性中只能选择两个：

Pipes的目标就是打破这个三角：同时提供 effectful（带效果）、streaming（流式）、composable（可组合）的编程模型。

二、两个原语与四类组件

为了强制实现松散耦合，Pipes 规定组件之间只能通过两条命令通信：

• yield：向下游发送输出数据；
• awaiting：从上游接收输入数据。

围绕这两条命令，Pipes 定义了四类组件（T后缀为更通用的 monad-transformer 版本）：

T后缀类型（ProducerT、ConsumerT、PipeT、EffectT）是更通用的 monad-transformer，可以把任意单子M提升进来，用M的行为补充 Pipes 的行为。非T后缀类型（Producer、Consumer、Pipe、Effect）只支持提升Eff<RT, A>，用起来更简单，但灵活性稍低。从 Pipe.cs 的注释可以确认：Pipe只提升Eff单子。

连接与运行

Pipes 使用**参数化多态（泛型）**重载所有操作。操作符|把Producer、Consumer、Pipe“融合（fuse）”在一起，最终融合成一个Effect或EffectT；只有融合成Effect/EffectT之后才能调用.Run()启动管线。Producer、ProducerT、Consumer、ConsumerT、Pipe、Effect、EffectT全都是同一个底层类型PipeT的特例。

三、统一底层类型PipeT

PipeT是所有 Pipes 组件的唯一底层形态，其形状如下：

PipeT<IN, OUT, M, A> Upstream | Downstream +---------+ | | IN --► --► OUT -- 信息向下游流动 | | | +----|----+ | A

源码定义位于 PipeT.cs：public abstract record PipeT<IN, OUT, M, A> : K<PipeT<IN, OUT, M>, A>，其中：

• IN：要消费的值的类型；
• OUT：要产生的值的类型；
• M：被提升的单子类型（约束为MonadIO<M>）；
• A：绑定值类型。

类型同义词：用Unit和Void封闭两端

Pipes 用类型同义词隐藏未使用的输入或输出，让类型签名更干净。类型同义词分两类：具体类型同义词（显式封闭Proxy未使用的输入/输出）和多态类型同义词（不显式封闭）。具体同义词用Unit封闭未使用的输入，用Void（无人居住的类型）封闭未使用的输出：

EffectT：显式封闭两端，禁止awaiting和yield

EffectT<M, A> = PipeT<Unit, Void, M, A> Upstream | Downstream +---------+ | | Unit --► --► Void | | | +----|----+ | A

源码对应：EffectT.cs 中EffectT<M, A>(PipeT<Unit, Void, M, A> Proxy)。

ProducerT：显式封闭上游端，禁止awaiting

ProducerT<OUT, M, A> = PipeT<Unit, OUT, M, A> Upstream | Downstream +---------+ | | Unit --► --► OUT | | | +----|----+ | A

源码对应：ProducerT.cs 中ProducerT<OUT, M, A>(PipeT<Unit, OUT, M, A> Proxy)。

ConsumerT：显式封闭下游端，禁止yield

ConsumerT<IN, M, A> = PipeT<IN, Void, M, A> Upstream | Downstream +---------+ | | IN --► --► Void | | | +----|----+ | A

源码对应：ConsumerT.cs 中ConsumerT<IN, M, A>(PipeT<IN, Void, M, A> Proxy)。

三种底层运行节点

PipeT的Run()语义由其内部记录类型决定（见 PipeT.DSL.cs）：

• PipeTPure（L44-L76）：Run()返回M.Pure(Value)，即把绑定值提升进单子；
• PipeTFail（L78-L110）：Run()返回M.Fail<A>(Value)，让错误沿Fallible通道传播；
• PipeTEmpty（L8-L42）：Run()返回M.Empty<A>()，空流直接结束；
• 而PipeTYield与PipeTAwait的Run()会抛出InvalidOperationException("closed")（L278-L280、L312-L314）——这正说明：未融合的 yield/await 节点不能直接运行，必须先用|融合成Effect/EffectT。

四、|操作符：横向融合

用|组合PipeT，本质上是把它们并排摆放并横向融合（fusing）。例如把一个ProducerT、一个PipeT、一个ConsumerT组合起来，信息流动如下：

ProducerT PipeT ConsumerT +-------------+ +------------+ +-------------+ | | | | | | Unit --► readLine --► string --► parseInt --► int --► writeLine --► Void | | | | | | | | | +------|------+ +------|-----+ +------|------+ | | | A A A

融合会把中间接口全部消除，只留下一个EffectT：

EffectT +-------------------------------------+ | | Unit --► readLine | parseInt | writeLine --► Void | | +------------------|------------------+ | A

这个EffectT可以Run()，返回融合后的底层M类型；如果是Effect，则返回融合后的底层Eff<RT, A>。

|的源码实现

|由类型上的操作符重载实现（PipeT.cs）：

public static PipeT<IN, OUT, M, A> operator | (PipeT<IN, OUT, M, A> lhs, PipeT<OUT, OUT, M, A> rhs) => lhs.Compose(rhs); public static ProducerT<OUT, M, A> operator | (ProducerT<IN, M, A> lhs, PipeT<IN, OUT, M, A> rhs) => lhs.Compose(rhs); public static ConsumerT<IN, M, A> operator | (PipeT<IN, OUT, M, A> lhs, ConsumerT<OUT, M, A> rhs) => lhs.Compose(rhs.Proxy);

可以看到类型约束是精确对齐的：相邻组件的输出类型必须等于下游组件的输入类型，这正是依赖类型系统保证管线类型正确的手段。ProducerT | ConsumerT则直接得到EffectT（ProducerT.cs）：

public static EffectT<M, A> operator | (ProducerT<OUT, M, A> lhs, ConsumerT<OUT, M, A> rhs) => lhs.Proxy.Compose(rhs.Proxy).ToEffect();

融合的底层机制是PipeT的四个内部抽象方法（PipeT.cs）：

• ReplaceAwait：用上游 producer 替换awaiting节点；
• ReplaceYield：用下游 consumer 替换yield节点；
• PairEachAwaitWithYield/PairEachYieldWithAwait：逐个配对相邻节点的请求与产出。

例如 PipeT.DSL.cs 中PipeTYield.ReplaceYield会调用下游consumer(Value)并继续融合后续节点，PipeTAwait.ReplaceAwait（L300-L302）则用producer().Bind(x => Await(x).ReplaceAwait(producer))递归替换——这就是“中间接口被融合消除”的代码级真相。

五、构建组件：DSL 构造器

PipeT.Module.cs 提供了所有组件共享的 DSL 构造器，非T版本则在各自的 Module 类中薄封装并固定M = Eff<RT>：

非T版本的封装可见 Producer.Module.cs：Producer.yield<RT, OUT>(value)内部就是PipeT.yield<Eff<RT>, Unit, OUT>(value)；Consumer.Module.cs 的Consumer.awaiting<RT, IN>()则是PipeT.awaiting<Eff<RT>, IN, Void>().ToConsumer()。这印证了“非 T 类型只是把M固定为Eff<RT>的薄封装”。

Effect与Schedule的便捷重载

融合完成后，Effect/EffectT还重载了与Schedule的|操作（Effect.cs、EffectT.cs），例如effect | schedule表示按计划重复执行整个管线（Schedule.Forever时无限重复）：

public static Effect<RT, A> operator |(Effect<RT, A> lhs, Schedule rhs) => ReferenceEquals(rhs, Schedule.Forever) ? Effect.repeat(lhs) : Effect.repeat(rhs, lhs);

六、运行管线：Run()

融合成Effect/EffectT后即可运行：

• Effect<RT, A>.Run()返回Eff<RT, A>（Effect.cs）；
• EffectT<M, A>.Run()返回K<M, A>，即底层单子M（EffectT.cs）。

Run()内部最终落到PipeT.Run()（PipeT.cs），由各内部节点的Run语义逐层求值。Effect的Run()结果为Eff<RT, A>，之后还需要交给 RT 运行时执行，例如仓库测试中的用法（PipesTests.cs）：

using var rt = Runtime.New(); (merge(yield<Runtime, int>(1), yield<Runtime, int>(1)) | awaiting<Runtime, int>().Map(ignore)) .Run().As() .Run(rt, EnvIO.New()) .Ignore();

七、实战示例：文本文件分块流

文档推荐的示例 TextFileChunkStreamExample.cs 完整演示了 Producer → Pipe → Consumer 的融合管线，它把文件内容按 40 字符分块流式输出：

static Effect<RT, Unit> mainEffect(string path) => File<RT>.openRead(path) | Stream<Eff<RT>>.read(80).ToEff() | decodeUtf8 | writeLine; static Pipe<RT, SeqLoan<byte>, string, Unit> decodeUtf8 => from c in Pipe.awaiting<RT, SeqLoan<byte>, string>() from _ in Pipe.yield<RT, SeqLoan<byte>, string>(Encoding.UTF8.GetString(c.ToReadOnlySpan())) select unit; static Consumer<RT, string, Unit> writeLine => from l in Consumer.awaiting<RT, string>() from _ in Console<RT>.writeLine(l) select unit;

逐段解读：

1. File<RT>.openRead(path)返回Producer<RT, Stream, Unit>，作为文件流的数据源（见 File.Eff.cs）；
2. Stream<Eff<RT>>.read(80).ToEff()是一个Pipe，把Stream切成SeqLoan<byte>块。底层实现（Stream.cs）先从ArrayPool<byte>.Shared租借 80 字节缓冲区异步读取，再用ToSeqLoanUnsafe复用缓冲池避免分配，读取完毕归还；.ToEff()来自 PipeT.Extensions.cs，把通用PipeT<IN, OUT, Eff<RT>, A>降为Pipe<RT, IN, OUT, A>；
3. decodeUtf8用 LINQ 查询语法（from/select，对应Bind/Map）实现Pipe组件：await 一块字节，yield 解码后的字符串；
4. writeLine是Consumer，await 一行并写到控制台。

注意整个mainEffect通过|把四个组件融合为Effect<RT, Unit>，全程没有显式 for 循环、没有一次性载入内存，效果（文件 IO、控制台 IO）被声明式地穿插在流式处理中——这正是文档所说“三者兼得”的直接体现。

八、更高阶的组合：merge与Conduit

Pipes 生态还提供了跨流组合能力。ProducerT.merge（ProducerT.Module.cs）可以把多个 producer 并行合并成一个：

public static ProducerT<OUT, M, Unit> merge<OUT, M>( Seq<ProducerT<OUT, M, Unit>> producers, Buffer<OUT>? settings = null) where M : MonadUnliftIO<M> { if (producers.Count == 0) return pure<OUT, M, Unit>(default); return from conduit in Pure(Conduit.make(settings ?? Buffer<OUT>.Unbounded)) from signal in Signal.countdown<M>(producers.Count) from forks in forkEffects(producers, signal, conduit) from _ in conduit.ToProducerT<M>() from x in forks.Traverse(f => f.Cancel).As() select unit; }

其实现依赖Conduit——一个“带内部队列的通道”（Conduit.cs）：sink 侧是反变（Comap/CoTransform），source 侧是协变（Map/Transform），内部队列由System.Threading.Channels.Channel支撑。Conduit.make支持多种缓冲策略（Conduit.Module.cs 与 Buffer.cs）：

通过Conduit.ToProducerT<T>()/ToConsumerT<T>()（Conduit.cs）可以把通道两端接入 Pipes 组件，实现跨线程的流式互通——这是 Pipes 统一机制之外的又一个扩展点。

九、总结

language-ext 的Pipes用“一个底层类型 + 两个原语 + 一个融合操作符”实现了流式、效果与可组合性的统一：

• 统一底层：ProducerT/ConsumerT/PipeT/EffectT都是PipeT<IN, OUT, M, A>的类型同义词，Unit/Void负责封闭未使用的端口；
• 两原语驱动：yield与awaiting是组件间仅有的两种通信方式，天然保证解耦；
• |横向融合：通过Compose与内部ReplaceAwait/ReplaceYield机制消除中间接口，最终产出可Run()的Effect/EffectT；
• 效果原生：M = Eff<RT>时可直接穿插文件、控制台等 IO 效果（参见 File.Eff.cs、Console.Eff.cs 中readLines等现成的Producer组件），且所有操作都在常量内存中完成。

想深入了解实现细节，建议按顺序阅读 PipeT.cs、PipeT.DSL.cs、PipeT.Module.cs 三个核心文件，再配合 PipesTests.cs 与 EffectsExamples 示例动手实验。

• 后端