企业官网建设流程全解析

• 后端
• 微服务

导读

Akka.NET Streams 是 Akka.NET 中基于 Actor 模型实现的响应式流（Reactive Streams）框架，用于以声明式、可组合的方式处理有界或无界数据流。本文以官方 Quickstart 指南为主线，带你完成从安装Akka.Streams包、创建第一个Source、运行流，到用Scan+FileIO计算阶乘并写入文件、用Flow封装可复用的Sink、用ZipWith与Throttle做基于时间的流控这一完整链路。读完本文，你将掌握 Akka Streams 的"蓝图（blueprint）"式编程模型、Materializer的作用、materialized value（物化值）的取舍规则，以及背压（back-pressure）为何能让十亿元素级别的流安全运行。

前提：添加 Akka.Streams 模块

Akka Streams 是独立于核心Akka之外的模块，使用时必须先将其加入项目。在包管理器控制台执行：

Install-Package Akka.Streams

安装完成后，在代码中导入两组命名空间，即可获得完整的流式编程工具集：

using Akka.Streams; using Akka.Streams.Dsl;

其中Akka.Streams提供了Source、Sink、Flow、Materializer、ThrottleMode等核心类型；Akka.Streams.Dsl则提供了构建流的领域专用语言（DSL），如FileIO、Keep以及大量组合算子。如果你想更系统地了解流式处理的背景概念，可以先阅读仓库中的 流式处理介绍 与 基础概念。

第一个 Source：描述而非执行

Akka Streams 中，一条流通常始于一个Source（数据源）。一个最简单的例子是发射 1 到 100 的整数：

Source<T, NotUsed> source = Source.From(Enumerable.Range(1, 100));

Source带有两个类型参数：

• 第一个参数（T）是这条流发射的元素类型；
• 第二个参数是该流被运行时产生的"辅助值"（auxiliary value）的类型，即materialized value（物化值）。例如网络型 Source 可能通过物化值暴露绑定的端口或对端地址；而一个简单的整数范围不产生任何辅助信息，因此使用NotUsed类型。

关键认知：创建Source只是得到了一个"如何发射前 100 个自然数"的蓝图（blueprint），它此刻并未激活，也没有任何元素被真正计算出来。要让数据真正流动，必须显式地"运行"它：

source.RunForeach(i => Console.WriteLine(i.ToString()), materializer);

这一行把消费函数（这里是打印到控制台）作为Sink与Source拼接，并将这条小流交给一个 Actor 去运行。Akka Streams 中所有会真正触发执行的 API 都遵循"名字里带 run"的命名惯例，例如RunForeach、RunWith、RunAggregate、RunSum等——看到 run 字样，就意味着流被激活了。从源码看，RunForeach的本质是把Sink.ForEach(action)与当前 Source 连接并运行（见 Source.cs），而RunWith则通过ToMaterialized(sink, Keep.Right).Run(materializer)完成拼接与运行（见 Source.cs）。

materializer 从哪里来

你可能好奇运行流的 Actor 是谁创建的、materializer又是什么。要获得这个值，首先需要创建一个 ActorSystem：

using (var system = ActorSystem.Create("system")) using (var materializer = system.Materializer()) { // 在这里运行你的流 }

Materializer是"流执行引擎的工厂"，它就是让流真正跑起来的那个东西。现阶段你不需要关心它的内部细节，只需要知道：调用Source上的任何 run 方法，都必须提供一个 materializer。

除了从ActorSystem创建，还有其他途径获得 materializer——例如在 Actor 内部使用流时，可以从ActorContext创建。仓库中的扩展方法签名证实了这一点（见 ActorMaterializer.cs）：

public static ActorMaterializer Materializer(this IActorRefFactory context, ActorMaterializerSettings settings = null, string namePrefix = null)

由于IActorRefFactory同时被ActorSystem与ActorContext实现，所以两种场景都能用同一个扩展方法拿到 materializer。此外，为了简化用法，许多 run 方法还提供了直接接收ActorSystem的重载（见 Source.cs 与 Source.cs），让你不必显式创建 materializer 变量。

蓝图复用：用 Scan 生成阶乘并写入文件

Akka Streams 最吸引人的特性之一：Source只是一份"你想运行什么"的描述，像建筑师的图纸一样，可以被复用、被嵌入到更大的设计中。下面我们把整数源变换为阶乘序列，再写入文件：

var factorials = source.Scan(new BigInteger(1), (acc, next) => acc * next); var result = factorials .Select(num => ByteString.FromString($"{num}\n")) .RunWith(FileIO.ToFile(new FileInfo("factorials.txt")), materializer);

这里发生了两件事：

1. Scan组合子：以BigInteger(1)为初始值，对流入的每个数依次做乘法累加。Scan会先发射初始值，随后发射每一次的计算结果，从而得到阶乘序列1, 1, 2, 6, 24, ...。注意Scan在 SourceOperations.cs 中的定义是Scan(flow, zero, Func<TOut2, TOut1, TOut2>)——它仍然只是对流的描述，此刻没有任何计算发生。
2. Select+FileIO.ToFile：把每个阶乘数字转成带换行符的ByteString，再把它接上文件接收端Sink并运行。

在 Akka Streams 术语中，数据流的接收端称为Sink。FileIO.ToFile返回的是Sink<ReadOnlySequence<byte>, Task<IOResult>>——它接收字节数据，并在物化时给出一个Task<IOResult>（见 FileIO.cs）。IOResult是 Akka Streams 中 IO 操作的统一返回类型，用于告知你：

• 流总共消费了多少字节（或元素）；
• 流是正常结束（Success）还是异常终止（Failure）。

result这个Task<IOResult>就可以用来等待文件写入完成、检查写入字节数或捕获异常。

Reusable Pieces：把整条处理链封装成 Flow

Akka Streams 另一个独到之处（也是其他许多流库不具备的）在于：不只 Source 可以像蓝图一样复用，流的其他元素同样可以。我们可以把"文件写入 Sink"和它前面的字节转换步骤打包成一个可复用组件。

由于流的描述语言始终从左向右书写（如同英文阅读顺序），我们需要一个"像 Source 但输入是开放的"起点——这就是Flow：

public static Sink<string, Task<IOResult>> LineSink(string filename) { return Flow.Create<string>() .Select(s => ByteString.FromString($"{s}\n")) .ToMaterialized(FileIO.ToFile(new FileInfo(filename)), Keep.Right); }

从字符串Flow出发，先把每个字符串转成ByteString，再喂给前面已经熟悉的文件写入Sink。得到的蓝图类型是Sink<string, Task<IOResult>>，含义是：它接收字符串作为输入，物化时会产生Task<IOResult>类型的辅助信息。

Keep.Right 与物化值取舍

在Source或Flow上链式拼接操作时，最终物化值的类型默认由最左侧起点决定。但这里我们想保留的是右侧FileIO.ToFile提供的Task<IOResult>（写入结果信息），所以必须显式声明Keep.Right。

Keep类提供了一组常用的物化值选择函数（见 Keep.cs）：

这也是Flow.ToMaterialized(sink, combine)中第二个参数的作用——combine正是这类Func<TMat, TMat2, TMat3>选择器（见 Flow.cs）。顺带一提，Flow.To(sink)默认等价于ToMaterialized(sink, Keep.Left)（见 Flow.cs）。

现在，把刚封装好的LineSink接到阶乘源上——只需要一个小小的适配，把数字转成字符串：

factorials.Select(_ => _.ToString()).RunWith(LineSink("factorial2.txt"), materializer);

基于时间的处理：ZipWith 与 Throttle

在进入更复杂的例子之前，先感受一下 Akka Streams 的"流式"本质。从factorials源出发，把它与另一条发射 0~100 整数的Source进行ZipWith组合：factorials发射的第一个元素是 0 的阶乘、第二个是 1 的阶乘，依此类推。二者组合成"3! = 6"这样的字符串：

await factorials .ZipWith(Source.From(Enumerable.Range(0, 100)), (num, idx) => $"{idx}! = {num}") .Throttle(1, TimeSpan.FromSeconds(1), 1, ThrottleMode.Shaping) .RunForeach(Console.WriteLine, materializer);

ZipWith在 SourceOperations.cs 中的签名是ZipWith(flow, other, Func<T1, T2, T3> combine)——它把两条流按位置一一配对，再用combine函数合成新元素。

前面所有操作都与时间无关，即便换成严格的元素集合也能照做。而Throttle证明我们处理的确实是"以一定速度流动"的流：它把流限速为每秒 1 个元素。

Throttle 的参数与两种模式

Throttle的完整签名（见 InternalFlowOperations.cs）：

public static IFlow<T, TMat> Throttle<T, TMat>( this IFlow<T, TMat> flow, int elements, // 速率：每个时间窗内允许的元素数 TimeSpan per, // 时间窗长度 int maximumBurst, // 允许的突发（burst）上限 ThrottleMode mode) // Shaping 或 Enforcing

在上面的例子中：

• elements = 1、per = 1 秒：限速为每秒 1 个元素；
• maximumBurst = 1：允许的最大突发量为 1——这意味着第一个元素立即通过，第二个元素必须等待 1 秒，第三个再等 1 秒，以此类推。

ThrottleMode定义了超速时的处理策略（见 ThrottleMode.cs）：

从实现上看，Throttle采用令牌桶（token bucket）模型：桶的容量即maximumBurst，令牌按固定速率滴入桶中、可积攒到桶容量以便容忍突发；流要发送元素时从桶中取令牌，取不到就等待（见 InternalFlowOperations.cs）。参数校验也有明确约束（见 InternalFlowOperations.cs）：

• elements必须大于 0，否则抛ArgumentException；
• per必须大于TimeSpan.Zero；
• 在Enforcing模式下，maximumBurst不能小于 0。

此外还有一个按"成本"限速的重载Throttle(cost, per, maximumBurst, calculateCost, mode)，允许你根据每个元素的不同代价（如消息大小）做加权限速（见 InternalFlowOperations.cs）。

运行上面的程序，你会看到每秒打印一行。但有一个不那么直观的要点值得注意：如果尝试把两条流都设置为产生十亿个数，你会发现环境不会因 OutOfMemoryError 崩溃——尽管流的运行是在后台异步进行的（这正是辅助信息以Task形式提供的原因）。

背压：Akka Streams 的秘密武器

流之所以能安全处理十亿级别的元素，秘密在于Akka Streams 隐式实现了全链路流量控制，所有组合子都尊重背压（back-pressure）。

Throttle会向上游所有数据源发出信号：它每秒钟只能接受一个元素；当上游注入速率高于每秒一个时，Throttle就会向上游施加背压。这个机制保证了：

• 上游不会一股脑地把元素全部推入内存；
• 每个阶段都按下游可消费的速率生产数据；
• 即使数据总量极大，内存占用也保持在稳定的小范围内。

这也是 Akka Streams 与普通 LINQ 集合操作的本质区别：前者是异步、带背压、逐元素流动的流，后者是同步、全量驻留内存的严格集合。

小结与下一步

总而言之，Akka Streams 的核心心智模型就三句话：

1. Source、Flow、Sink都是可复用的蓝图（描述），组合它们不会触发任何计算；
2. 调用名字中带run的方法（并提供Materializer）才会真正执行流；
3. 所有阶段之间自动施加背压，让流天然安全、可扩展。

以上只是 Akka Streams 的冰山一角——仓库中还提供了数十种内置 Source 与 Sink（文件、TCP、HTTP、队列、Actor 发布订阅等）以及更多流变换组合子（Map、Filter、Buffer、Fold、Grouped、Merge、Broadcast等）。你可以继续阅读 内置阶段的概览与语义 深入了解；如果想要动手实践更多组合方式，还可以参考 Streams 示例项目，以及 Quickstart 原文 对应的完整示例代码。

参考源码索引

• Source.cs：Source核心类型，RunWith、RunForeach等运行方法及ActorSystem重载；
• SourceOperations.cs：Scan、ZipWith、Throttle等扩展算子；
• Flow.cs：Flow.Create、To、ToMaterialized及物化值组合；
• FileIO.cs：FileIO.ToFile/FromFile与IOResult；
• Keep.cs：Keep.Left/Right/Both/None物化值选择函数；
• ThrottleMode.cs：Shaping/Enforcing两种限速策略；
• InternalFlowOperations.cs：Throttle令牌桶实现与参数校验；
• ActorMaterializer.cs：Materializer()扩展方法，从IActorRefFactory（ActorSystem/ActorContext）创建执行引擎。

• 后端
• 微服务