后端【免费下载链接】language-extC# pure functional programming framework - come and get declarative!项目地址https://gitcode.com/gh_mirrors/la/language-ext点击查看免费下载导读Pipes是 language-ext 中用于纯函数式流式编程的核心抽象它以单个统一的PipeT类型为基础派生出Producer、Consumer、Pipe、Effect四类可相互连接的流处理组件让你在常量内存中边流式处理边穿插副作用Effects同时保持组件级可组合性。读完本文你将理解 Pipes 的“三选二”困局解法、yield/awaiting两个原语、|操作符的横向融合原理、Run()的启动方式并能用仓库内真实示例文本文件分块流搭建自己的管线。这是一个相对复杂的概念初次接触感到困惑是正常的。仓库中提供了一个完整可运行的示例Samples/EffectsExamples/Examples/TextFileChunkStreamExample.cs建议配合本文一起阅读。一、传统流编程的“三选二”困局先看文档指出的核心痛点常规的流编程迫使你在以下三个特性中只能选择两个牺牲项得到的方案代价牺牲EffectsIEnumerable可用可组合的函数在常量空间内变换但无法穿插副作用除非是命令式的那种牺牲StreamingTraverse/Sequence可组合、可带效果但必须等整个列表先处理完并载入内存才能返回单个结果牺牲Composability紧耦合的 for 循环边流式处理边触发命令式副作用但不可模块化、不可分离Pipes的目标就是打破这个三角同时提供 effectful带效果、streaming流式、composable可组合的编程模型。二、两个原语与四类组件为了强制实现松散耦合Pipes 规定组件之间只能通过两条命令通信yield向下游发送输出数据awaiting从上游接收输入数据。围绕这两条命令Pipes 定义了四类组件T后缀为更通用的 monad-transformer 版本组件能力可用操作Producer/ProducerT只向下游 yieldProducer.yield、ProducerT.yieldConsumer/ConsumerT只从上游 awaitingConsumer.awaiting、ConsumerT.awaitingPipe/PipeT既能 await 又能 yieldPipe.awaiting、PipeT.awaiting、Pipe.yield、PipeT.yieldEffect/EffectT既不 yield 也不 await建模非流式组件只有绑定值AT后缀类型ProducerT、ConsumerT、PipeT、EffectT是更通用的 monad-transformer可以把任意单子M提升进来用M的行为补充 Pipes 的行为。非T后缀类型Producer、Consumer、Pipe、Effect只支持提升EffRT, A用起来更简单但灵活性稍低。从 Pipe.cs 的注释可以确认Pipe只提升Eff单子。连接与运行Pipes 使用**参数化多态泛型**重载所有操作。操作符|把Producer、Consumer、Pipe“融合fuse”在一起最终融合成一个Effect或EffectT只有融合成Effect/EffectT之后才能调用.Run()启动管线。Producer、ProducerT、Consumer、ConsumerT、Pipe、Effect、EffectT全都是同一个底层类型PipeT的特例。三、统一底层类型PipeTPipeT是所有 Pipes 组件的唯一底层形态其形状如下PipeTIN, OUT, M, A Upstream | Downstream --------- | | IN --► --► OUT -- 信息向下游流动 | | | ----|---- | A源码定义位于 PipeT.cspublic abstract record PipeTIN, OUT, M, A : KPipeTIN, OUT, M, A其中IN要消费的值的类型OUT要产生的值的类型M被提升的单子类型约束为MonadIOMA绑定值类型。类型同义词用Unit和Void封闭两端Pipes 用类型同义词隐藏未使用的输入或输出让类型签名更干净。类型同义词分两类具体类型同义词显式封闭Proxy未使用的输入/输出和多态类型同义词不显式封闭。具体同义词用Unit封闭未使用的输入用Void无人居住的类型封闭未使用的输出EffectT显式封闭两端禁止awaiting和yieldEffectTM, A PipeTUnit, Void, M, A Upstream | Downstream --------- | | Unit --► --► Void | | | ----|---- | A源码对应EffectT.cs 中EffectTM, A(PipeTUnit, Void, M, A Proxy)。ProducerT显式封闭上游端禁止awaitingProducerTOUT, M, A PipeTUnit, OUT, M, A Upstream | Downstream --------- | | Unit --► --► OUT | | | ----|---- | A源码对应ProducerT.cs 中ProducerTOUT, M, A(PipeTUnit, OUT, M, A Proxy)。ConsumerT显式封闭下游端禁止yieldConsumerTIN, M, A PipeTIN, Void, M, A Upstream | Downstream --------- | | IN --► --► Void | | | ----|---- | A源码对应ConsumerT.cs 中ConsumerTIN, M, A(PipeTIN, Void, M, A Proxy)。三种底层运行节点PipeT的Run()语义由其内部记录类型决定见 PipeT.DSL.csPipeTPureL44-L76Run()返回M.Pure(Value)即把绑定值提升进单子PipeTFailL78-L110Run()返回M.FailA(Value)让错误沿Fallible通道传播PipeTEmptyL8-L42Run()返回M.EmptyA()空流直接结束而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融合会把中间接口全部消除只留下一个EffectTEffectT ------------------------------------- | | Unit --► readLine | parseInt | writeLine --► Void | | ------------------|------------------ | A这个EffectT可以Run()返回融合后的底层M类型如果是Effect则返回融合后的底层EffRT, A。|的源码实现|由类型上的操作符重载实现PipeT.cspublic static PipeTIN, OUT, M, A operator | (PipeTIN, OUT, M, A lhs, PipeTOUT, OUT, M, A rhs) lhs.Compose(rhs); public static ProducerTOUT, M, A operator | (ProducerTIN, M, A lhs, PipeTIN, OUT, M, A rhs) lhs.Compose(rhs); public static ConsumerTIN, M, A operator | (PipeTIN, OUT, M, A lhs, ConsumerTOUT, M, A rhs) lhs.Compose(rhs.Proxy);可以看到类型约束是精确对齐的相邻组件的输出类型必须等于下游组件的输入类型这正是依赖类型系统保证管线类型正确的手段。ProducerT | ConsumerT则直接得到EffectTProducerT.cspublic static EffectTM, A operator | (ProducerTOUT, M, A lhs, ConsumerTOUT, M, A rhs) lhs.Proxy.Compose(rhs.Proxy).ToEffect();融合的底层机制是PipeT的四个内部抽象方法PipeT.csReplaceAwait用上游 producer 替换awaiting节点ReplaceYield用下游 consumer 替换yield节点PairEachAwaitWithYield/PairEachYieldWithAwait逐个配对相邻节点的请求与产出。例如 PipeT.DSL.cs 中PipeTYield.ReplaceYield会调用下游consumer(Value)并继续融合后续节点PipeTAwait.ReplaceAwaitL300-L302则用producer().Bind(x Await(x).ReplaceAwait(producer))递归替换——这就是“中间接口被融合消除”的代码级真相。五、构建组件DSL 构造器PipeT.Module.cs 提供了所有组件共享的 DSL 构造器非T版本则在各自的 Module 类中薄封装并固定M EffRT构造器语义源码位置yield(value)向下游产出单个值PipeT.Module.cs#L22-L24yieldAll(xs)依次向下游产出IEnumerable/IAsyncEnumerable/Source/SourceT中的全部值PipeT.Module.cs#L33-L105yieldRepeat(ma)反复执行单子ma并向下游产出其绑定值PipeT.Module.cs#L114-L116awaiting()从上游等待一个值PipeT.Module.cs#L136-L138map(f)用映射函数构造一个 Pipe先 await 再 yieldPipeT.Module.cs#L159-L161mapM(f)映射函数返回单子值边处理边穿插效果PipeT.Module.cs#L171-L173filter(pred)过滤不满足谓词的值PipeT.Module.cs#L147-L149pure(value)/empty()/error(e)/fail(e)只返回绑定值 / 空流 / 以Error失败 / 以任意E失败PipeT.Module.cs#L183-L222lift(f)/liftT(f)/liftM(ma)/liftIO(ma)提升纯函数、惰性 Pipe、单子、IOPipeT.Module.cs#L232-L330repeat(ma)/repeat(schedule, ma)无限重复 / 按Schedule重复PipeT.Module.cs#L340-L365fold/foldUntil/foldWhile折叠流按Schedule或谓词决定何时把聚合值 yield 出去PipeT.Module.cs#L404-L692非T版本的封装可见 Producer.Module.csProducer.yieldRT, OUT(value)内部就是PipeT.yieldEffRT, Unit, OUT(value)Consumer.Module.cs 的Consumer.awaitingRT, IN()则是PipeT.awaitingEffRT, IN, Void().ToConsumer()。这印证了“非 T 类型只是把M固定为EffRT的薄封装”。Effect与Schedule的便捷重载融合完成后Effect/EffectT还重载了与Schedule的|操作Effect.cs、EffectT.cs例如effect | schedule表示按计划重复执行整个管线Schedule.Forever时无限重复public static EffectRT, A operator |(EffectRT, A lhs, Schedule rhs) ReferenceEquals(rhs, Schedule.Forever) ? Effect.repeat(lhs) : Effect.repeat(rhs, lhs);六、运行管线Run()融合成Effect/EffectT后即可运行EffectRT, A.Run()返回EffRT, AEffect.csEffectTM, A.Run()返回KM, A即底层单子MEffectT.cs。Run()内部最终落到PipeT.Run()PipeT.cs由各内部节点的Run语义逐层求值。Effect的Run()结果为EffRT, A之后还需要交给 RT 运行时执行例如仓库测试中的用法PipesTests.csusing var rt Runtime.New(); (merge(yieldRuntime, int(1), yieldRuntime, int(1)) | awaitingRuntime, int().Map(ignore)) .Run().As() .Run(rt, EnvIO.New()) .Ignore();七、实战示例文本文件分块流文档推荐的示例 TextFileChunkStreamExample.cs 完整演示了 Producer → Pipe → Consumer 的融合管线它把文件内容按 40 字符分块流式输出static EffectRT, Unit mainEffect(string path) FileRT.openRead(path) | StreamEffRT.read(80).ToEff() | decodeUtf8 | writeLine; static PipeRT, SeqLoanbyte, string, Unit decodeUtf8 from c in Pipe.awaitingRT, SeqLoanbyte, string() from _ in Pipe.yieldRT, SeqLoanbyte, string(Encoding.UTF8.GetString(c.ToReadOnlySpan())) select unit; static ConsumerRT, string, Unit writeLine from l in Consumer.awaitingRT, string() from _ in ConsoleRT.writeLine(l) select unit;逐段解读FileRT.openRead(path)返回ProducerRT, Stream, Unit作为文件流的数据源见 File.Eff.csStreamEffRT.read(80).ToEff()是一个Pipe把Stream切成SeqLoanbyte块。底层实现Stream.cs先从ArrayPoolbyte.Shared租借 80 字节缓冲区异步读取再用ToSeqLoanUnsafe复用缓冲池避免分配读取完毕归还.ToEff()来自 PipeT.Extensions.cs把通用PipeTIN, OUT, EffRT, A降为PipeRT, IN, OUT, AdecodeUtf8用 LINQ 查询语法from/select对应Bind/Map实现Pipe组件await 一块字节yield 解码后的字符串writeLine是Consumerawait 一行并写到控制台。注意整个mainEffect通过|把四个组件融合为EffectRT, Unit全程没有显式 for 循环、没有一次性载入内存效果文件 IO、控制台 IO被声明式地穿插在流式处理中——这正是文档所说“三者兼得”的直接体现。八、更高阶的组合merge与ConduitPipes 生态还提供了跨流组合能力。ProducerT.mergeProducerT.Module.cs可以把多个 producer 并行合并成一个public static ProducerTOUT, M, Unit mergeOUT, M( SeqProducerTOUT, M, Unit producers, BufferOUT? settings null) where M : MonadUnliftIOM { if (producers.Count 0) return pureOUT, M, Unit(default); return from conduit in Pure(Conduit.make(settings ?? BufferOUT.Unbounded)) from signal in Signal.countdownM(producers.Count) from forks in forkEffects(producers, signal, conduit) from _ in conduit.ToProducerTM() from x in forks.Traverse(f f.Cancel).As() select unit; }其实现依赖Conduit——一个“带内部队列的通道”Conduit.cssink 侧是反变Comap/CoTransformsource 侧是协变Map/Transform内部队列由System.Threading.Channels.Channel支撑。Conduit.make支持多种缓冲策略Conduit.Module.cs 与 Buffer.csBufferA策略行为Channel 配置BufferA.Unbounded默认无界队列CreateUnboundedBufferA.Bounded(size)有界队列满则等待BoundedChannelOptions(size)FullMode.WaitBufferA.Single()容量 1满则等待容量 1 WaitBufferA.Latest(initial)只保留最新值以initial起步永不满不空容量 1 DropOldestBufferA.Newest(size)保留最新size个溢出丢弃最旧容量sizeDropOldestBufferA.New()只保留最新一个容量 1 DropOldest通过Conduit.ToProducerTT()/ToConsumerTT()Conduit.cs可以把通道两端接入 Pipes 组件实现跨线程的流式互通——这是 Pipes 统一机制之外的又一个扩展点。九、总结language-ext 的Pipes用“一个底层类型 两个原语 一个融合操作符”实现了流式、效果与可组合性的统一统一底层ProducerT/ConsumerT/PipeT/EffectT都是PipeTIN, OUT, M, A的类型同义词Unit/Void负责封闭未使用的端口两原语驱动yield与awaiting是组件间仅有的两种通信方式天然保证解耦|横向融合通过Compose与内部ReplaceAwait/ReplaceYield机制消除中间接口最终产出可Run()的Effect/EffectT效果原生M EffRT时可直接穿插文件、控制台等 IO 效果参见 File.Eff.cs、Console.Eff.cs 中readLines等现成的Producer组件且所有操作都在常量内存中完成。想深入了解实现细节建议按顺序阅读 PipeT.cs、PipeT.DSL.cs、PipeT.Module.cs 三个核心文件再配合 PipesTests.cs 与 EffectsExamples 示例动手实验。赞分享后端【免费下载链接】language-extC# pure functional programming framework - come and get declarative!项目地址https://gitcode.com/gh_mirrors/la/language-ext点击查看免费下载相关推荐language-ext Effects 深度指南IO、Eff 与 EffRT, A 三大副作用封装单子全解析language ext Effects 深度指南IO、Eff 与 EffRT, A 三大副作用封装单子全解析 本文基于 LanguageExt.Core/后端Nintendo Switch特斯拉菜单终极指南如何快速配置Tesla-Menu覆盖菜单Nintendo Switch特斯拉菜单终极指南如何快速配置Tesla Menu覆盖菜单 Tesla Menu是Nintendo Switch自制系统中不可或嵌入式免费开源小说阅读器终极指南5分钟快速上手Uncle小说PC版免费开源小说阅读器终极指南5分钟快速上手Uncle小说PC版 还在为找不到心仪小说而烦恼吗Uncle小说PC版是你的一站式解决方案这款免费开源的小说下载阅桌面应用创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考