Adobe 的实时消息传递协议（RTMP），是一种应用层协议，旨在通过适当的传输协议（如 TCP）多路复用和打包多媒体传输流（如音频、视频和交互内容）。以下是对 RTMP 规范的详细解析。

文档概述

文件状态

本文档是 2012 年 12 月 21 日发布的 “Adobe 的实时消息传递协议” 规范的机器可读版本。自 2012 年 PDF 版本以来，规范内容并未发生实质性变化，仅在格式和文字编辑上有所调整。

引言

RTMP 提供了在可靠的流传输（如 TCP）上的双向消息多路复用服务，旨在携带视频、音频和数据消息的并行流，并附带相关的时间信息。实现通常会为不同类别的消息分配不同的优先级，这会影响在传输容量受限时消息入队到底层流传输的顺序。

术语

• MUST: 必须
• REQUIRED: 必需
• SHALL: 应该
• SHOULD: 建议
• MAY: 可以

贡献者

• Rajesh Mallipeddi：原 Adobe Systems 编辑，提供了大部分原始文本。
• Mohit Srivastava：Adobe Systems 贡献者。

定义

• Payload: 包含在数据包中的数据，如音频样本或压缩视频数据。
• Packet: 由固定头和负载数据组成的数据包。
• Port: 传输协议用来区分主机内多个目的地的抽象。
• Transport address: 用于识别传输层端点的网络地址和端口的组合。

字节顺序、对齐和时间格式

• 所有整数字段按网络字节顺序（大端序）传输。
• 时间戳以毫秒为单位，相对于一个未指定的纪元。

RTMP 块流

消息格式

消息格式应包含以下必要字段：

• Timestamp: 消息的时间戳（4 字节）。
• Length: 消息负载的长度（3 字节）。
• Type ID: 消息类型 ID（1 字节）。
• Message Stream ID: 消息流 ID（4 字节，小端序）。

握手

RTMP 连接从握手开始，客户端和服务器各发送相同的三个块（C0、C1、C2 和 S0、S1、S2）。

块化

在握手后，连接多路复用一个或多个块流。每个块流携带一种类型的消息。每个块都有唯一的块流 ID。块的传输顺序必须完整发送。接收端根据块流 ID 重新组装消息。

块格式

每个块由一个头和数据组成。头有三个部分：

• Basic Header: 编码块流 ID 和块类型（1-3 字节）。
• Message Header: 编码关于消息的信息（0、3、7 或 11 字节）。
• Extended Timestamp: 编码完整的 32 位时间戳（0 或 4 字节）。

RTMP 消息格式

消息头

消息头包含以下字段：

• Message Type: 消息类型（1 字节）。
• Length: 负载大小（3 字节）。
• Timestamp: 消息的时间戳（4 字节）。
• Message Stream ID: 消息流 ID（3 字节）。

用户控制消息

RTMP 使用消息类型 ID 4 进行用户控制消息。这些消息包含 RTMP 流层使用的信息。

RTMP 消息类型

命令消息

携带客户端和服务器之间的 AMF 编码命令。命令消息有两个类型值：20 表示 AMF0 编码，17 表示 AMF3 编码。

数据消息

用于发送元数据或用户数据。类型值为 18（AMF0）和 15（AMF3）。

共享对象消息

用于管理多个客户端和服务器之间的分布式数据。类型值为 19（AMF0）和 16（AMF3）。

音频消息

用于发送音频数据，类型值为 8。

视频消息

用于发送视频数据，类型值为 9。

聚合消息

包含一系列RTMP 子消息的单一消息。类型值为 22。

消息交换示例

发布录制视频

此示例说明发布者如何发布流并将视频流发送到服务器。其他客户端可以订阅此发布的流并播放视频。

+--------------------+                     +-----------+
|  Publisher Client  |        |            |   Server  |
+----------+---------+        |            +-----+-----+
          |           Handshaking Done           |
          |                  |                  |
          |                  |                  |
---+---- |----- Command Message(connect) ----->|
   |     |                                     |
   |     |<----- Window Acknowledge Size ------|
Connect |     |                                     |
   |     |<------ Set Peer BandWidth ----------|
   |     |                                     |
   |     |------ Window Acknowledge Size ----->|
   |     |                                     |
   |     |<----- User Control(StreamBegin) ----|
   |     |                                     |
---+---- |<-------- Command Message -----------|
          |   (_result- connect response)       |
          |                                     |
---+---- |--- Command Message(createStream) -->|
Create |     |                                     |
Stream |     |                                     |
---+---- |<------- Command Message ------------|
          | (_result- createStream response)    |
          |                                     |
---+---- |---- Command Message(publish) ------>|
   |     |                                     |
   |     |<----- User Control(StreamBegin) ----|
   |     |                                     |
   |     |---- Data Message (Metadata) ------->|
Publishing|     |                                     |
Content   |     |------------ Audio Data ------------>|
   |     |                                     |
   |     |------------ SetChunkSize ---------->|
   |     |                                     |
   |     |<--------- Command Message ----------|
   |     |      (_result- publish result)      |
   |     |                                     |
   |     |------------- Video Data ----------->|
   |     |                  |                  |
   |     |                  |                  |
          |    Until the stream is complete     |
          |                  |                  |

广播共享对象消息

此示例说明在创建和更改共享对象期间交换的消息。它还说明了共享对象消息广播的过程。

+----------+                       +----------+
|  Client  |           |           |  Server  |
+-----+----+           |           +-----+----+
       |   Handshaking and Application    |
       |          connect done            |
       |                |                 |
       |                |                 |
       |                |                 |
       |                |                 |
Create and ---+---- |---- Shared Object Event(Use)---->|
connect       |     |                                  |
Shared Object |     |                                  |
---+---- |<---- Shared Object Event --------|
       |       (UseSuccess,Clear)         |
       |                                  |
---+---- |------ Shared Object Event ------>|
Shared object |     |         (RequestChange)          |
Set Property  |     |                                  |
---+---- |<------ Shared Object Event ------|
       |            (Success)             |
       |                                  |
---+---- |------- Shared Object Event ----->|
Shared object|     |           (SendMessage)          |
Message      |     |                                  |
Broadcast ---+---- |<------- Shared Object Event -----|
       |           (SendMessage)          |
                          |                 |
                          |                 |

从录制流发布元数据

此示例描述了发布元数据的消息交换。

+------------------+                       +---------+
| Publisher Client |         |             |   FMS   |
+---------+--------+         |             +----+----+
       |     Handshaking and Application     |
       |            connect done             |
       |                  |                  |
       |                  |                  |
---+--- |-- Command Messsage (createStream) ->|
Create |    |                                     |
Stream |    |                                     |
---+--- |<-------- Command Message -----------|
       |   (_result - command response)      |
       |                                     |
---+--- |---- Command Message (publish) ----->|
Publishing |    |                                     |
metadata |    |<----- UserControl (StreamBegin) ----|
from file |    |                                     |
          |    |---- Data Message (Metadata) ------->|
       |                                     |

参考文献

• RFC0791: Postel, J., “Internet Protocol”, STD 5, RFC 791, September 1981.
• RFC0793: Postel, J., “Transmission Control Protocol”, STD 7, RFC 793, September 1981.
• RFC1982: Elz, R. and R. Bush, “Serial Number Arithmetic”, RFC 1982, August 199

在Ubuntu下使用apt安装MySQL后，root用户的密码默认是空的。这是因为在安装过程中，MySQL会自动为root用户设置一个随机密码，并将其存储在/etc/mysql/debian.cnf文件中[1]。

要解决这个问题，你可以按照以下步骤操作：

1. 使用以下命令查看MySQL自动设置的随机账户和密码：

   sudo cat /etc/mysql/debian.cnf

这将显示默认的随机账户和密码，你可以使用这组账户和密码登录MySQL[1]。

2. 如果你想修改root用户的密码，可以按照以下步骤操作：

• 登录MySQL： mysql -uroot -p 这将要求你输入密码，直接按回车键即可登录，因为默认密码为空。
• 切换到mysql数据库： use mysql;
• 修改root用户的密码： ALTER USER 'root'@'localhost' IDENTIFIED WITH mysql_native_password BY 'your_new_password'; 将your_new_password替换为你想要设置的新密码。
• 刷新权限： FLUSH PRIVILEGES;
• 退出MySQL： exit; 现在你可以使用新密码登录root用户了。

请注意，如果你使用的是MySQL 8.0版本，修改root用户密码的命令略有不同，请参考[1]。

Learn more:

1. Ubunto20.04安装MySQL并修改root用户密码（Linux安装mysql root用户无法登陆）_ubuntu20.04安装mysql设置root账号密码-CSDN博客
2. Ubuntu 20.4安装 mysql 8.0.20 以及解决root用户无法登录问题_ubuntu mysql 设置root密码后进不了-CSDN博客
3. Ubunto20.04安装MySQL并修改root用户密码（Linux安装mysql root用户无法登陆） – 青山是谁 – 博客园

RTMP协议（Real-Time Messaging Protocol）是一种用于在互联网上流式传输音频、视频和数据的通信协议[1]。最初由Macromedia开发为Flash Player和Flash Communication Server之间的流媒体传输的专有协议，后来被Adobe（收购了Macromedia）发布了该协议的不完整版本供公众使用[1]。

RTMP协议有多个变种：

• RTMP：在Transmission Control Protocol（TCP）之上工作的“纯”协议，默认使用1935端口号。
• RTMPS：在Transport Layer Security（TLS/SSL）连接上的RTMP。
• RTMPE：使用Adobe自己的安全机制对RTMP进行加密。虽然实现的细节是专有的，但该机制使用行业标准的加密原语[1]。
• RTMPT：封装在HTTP请求中以穿越防火墙。RTMPT通常使用TCP端口80和443上的明文请求来绕过大多数企业流量过滤。封装的会话可以在内部携带纯RTMP、RTMPS或RTMPE数据包[1]。
• RTMFP：在User Datagram Protocol（UDP）上的RTMP，取代了RTMP Chunk Stream。Secure Real-Time Media Flow Protocol套件由Adobe Systems开发，使终端用户能够直接连接和通信（P2P）[1]。

RTMP的基本操作是基于TCP的，它维持持久连接并允许低延迟通信。为了平稳地传输流并尽可能多地传输信息，它将流分割成片段，并且片段的大小在客户端和服务器之间动态协商。有时，片段的大小保持不变；音频数据的默认片段大小为64字节，视频数据和大多数其他数据类型的默认片段大小为128字节。来自不同流的片段可以交错和多路复用到单个连接上。由于较长的数据块，该协议每个片段仅携带一个字节的头部，因此开销非常小。然而，在实践中，通常不会交错单个片段。相反，交错和多路复用是在数据包级别上完成的，以确保每个通道满足其带宽、延迟和其他服务质量要求。以这种方式交错的数据包被视为不可分割的，不会在片段级别上交错[1]。

RTMP定义了几个虚拟通道，可以在这些通道上发送和接收数据包，并且这些通道彼此独立运行。例如，有一个用于处理RPC请求和响应的通道，一个用于视频流数据的通道，一个用于音频流数据的通道，一个用于带外控制消息（片段大小协商等）的通道等。在典型的RTMP会话期间，多个通道可能同时处于活动状态。当对RTMP数据进行编码时，会生成一个数据包头部。数据包头部指定了要发送的通道的ID、生成时间戳（如果需要）以及数据包有效载荷的大小。然后，在将其发送到连接上之前，将其实际有效载荷内容（媒体数据）根据当前协商的片段大小进行分段。数据包头部本身永远不会被分段，其大小不计入数据包的第一个片段中的数据。换句话说，只有实际的数据包有效载荷（媒体数据）会被分段[1]。

在更高的层次上，RTMP封装了MP3或AAC音频和FLV1视频多媒体流，并可以使用Action Message Format进行远程过程调用（RPC）。所需的任何RPC服务都是异步进行的，使用单个客户端/服务器请求/响应模型，因此不需要实时通信[2]。

RTMP会话可以使用以下两种方法进行加密：

• 使用行业标准的TLS/SSL机制。底层的RTMP会话只是包装在正常的TLS/SSL会话中。
• 使用RTMPE，在RTMP会话中包装一个轻量级的加密层。

在RTMP Tunneled（RTMPT）中，RTMP数据通过HTTP进行封装和交换，客户端（媒体播放器）的消息被发送到服务器上的80端口（HTTP的默认端口）[1]。RTMPT中的消息由于HTTP头部的原因比等效的非隧道化RTMP消息要大，但在客户端位于阻止非HTTP和非HTTPS出站流量的防火墙后，RTMPT可能有助于使用RTMP的场景，否则将无法使用非隧道化的RTMP。

RTMP协议的数据包结构如下：

• 数据包通过TCP连接发送，首先在客户端和服务器之间建立连接。
• 数据包包含一个头部和一个主体，在连接和控制命令的情况下，主体使用Action Message Format（AMF）进行编码。
• 头部分为基本头部（从图表中分离出来）和块消息头部。基本头部是数据包的唯一固定部分，通常由一个复合字节组成，其中最高的两个有效位是块类型（在规范中称为fmt），其余部分形成流ID。根据前者的值，可以省略消息头部的某些字段，并且可以从先前的数据包中派生它们的值，而根据后者的值，基本头部可以扩展为一个或两个额外的字节。如果基本头部的剩余六个位（最低有效位）的值为0，则基本头部为两个字节，表示从流ID 64到319（64+255）；如果值为1，则基本头部为三个字节（最后两个字节编码为16位小端），表示从流ID 64到65599（64+65535）；如果值为2，则基本头部为一个字节，保留用于低级协议控制消息和命令。块消息头部包含元数据信息，例如消息大小（以字节为单位）、时间戳增量和消息类型。最后一个值是一个字节，定义数据包是音频、视频、命令还是“低级”RTMP数据包，例如RTMP Ping[1]。

Learn more:

1. Real-Time Messaging Protocol – Wikipedia
2. RTMP协议深度解析：从原理到实践，掌握实时流媒体传输技术-阿里云开发者社区
3. 实时消息协议 – 维基百科，自由的百科全书

RTP（Real-time Transport Protocol）是一种用于实时流媒体传输的协议。它是一种面向数据报的协议，用于在IP网络上传输音频和视频等实时数据。RTP协议通常与RTCP（Real-time Transport Control Protocol）一起使用，RTCP用于传输控制信息，例如流媒体的质量反馈和同步信息。

RTP协议的特点和功能包括：

1. 实时传输：RTP协议被设计用于实时传输，可以在网络上以实时的方式传输音频和视频数据。
2. 分组传输：RTP将音频和视频数据分割成小的数据包进行传输，每个数据包都带有序列号和时间戳等信息，以便接收端可以正确地重建数据流。
3. 时间同步：RTP协议使用时间戳来确保音频和视频数据在接收端能够按照正确的时间顺序播放。
4. 负载类型：RTP协议支持多种不同的负载类型，例如音频编码（如G.711、AAC）和视频编码（如H.264、VP8）等。
5. QoS支持：RTP协议可以通过设置不同的服务质量（QoS）参数来满足实时流媒体传输的要求，例如延迟、带宽和抖动等。

RTP协议在实时流媒体传输中起着重要的作用，它可以与其他协议结合使用，例如RTSP（Real-time Streaming Protocol）用于控制流媒体的会话和传输。

Learn more:

1. 实时传输 Web 音频与视频 – Web 媒体技术 | MDN
2. 【RTP 传输协议】实时视频传输的艺术：深入探索 RTP 协议及其在 C++ 中的实现-阿里云开发者社区
3. RTP协议详解-CSDN博客

Koa 是由 Express 的原始团队开发的，旨在成为一个更小、更富表现力且更健壮的 Web 框架。Koa2 是 Koa 的第二个主要版本，具有一些显著的改进和特性。以下是对 Koa2 框架的详细介绍和分析：

Koa2 的特点

1. 轻量和模块化：
   • Koa 本身非常轻量，没有内置的中间件。
   • 它鼓励用户通过模块化的方式来构建应用，只需安装需要的中间件即可。
2. 现代 JavaScript 支持：
   • Koa2 采用了 async/await 语法，取代了 Koa1 中的 generator 函数。这使得异步代码更加简洁和易读。
   • 完全支持 ES6/ES7+ 特性，利用了现代 JavaScript 的优势。
3. 中间件机制：
   • Koa 的中间件机制类似于洋葱模型，支持中间件的顺序执行和嵌套。
   • 中间件可以在请求处理前后执行代码，提供了极大的灵活性。
4. 无内置路由：
   • 与 Express 不同，Koa2 不包括内置的路由功能。用户需要选择第三方路由库，如 koa-router。
5. 更好的错误处理：
   • Koa2 通过 try/catch 块来处理异步错误，避免了传统的回调地狱和难以调试的问题。

Koa2 的使用场景

1. API 服务：
   • 由于其轻量和高性能，Koa2 非常适合用于构建 RESTful API 服务。
   • 中间件机制可以轻松处理身份验证、日志记录、错误处理等任务。
2. 单页应用（SPA）后端：
   • Koa2 可以用作前端单页应用的后端，提供数据接口和简单的路由功能。
3. 微服务架构：
   • 由于其模块化设计，Koa2 非常适合于微服务架构中的单个服务。
   • 每个微服务都可以独立部署和扩展。

Koa2 示例代码

以下是一个简单的 Koa2 应用示例：

const Koa = require('koa');
const Router = require('koa-router');

const app = new Koa();
const router = new Router();

// 中间件示例
app.use(async (ctx, next) => {
  console.log(`{ctx.url}`);
  await next();
});

// 路由示例
router.get('/', async (ctx, next) => {
  ctx.body = 'Hello, Koa2!';
});

router.get('/hello/:name', async (ctx, next) => {
  const name = ctx.params.name;
  ctx.body = `Hello, ${name}!`;
});

app
  .use(router.routes())
  .use(router.allowedMethods());

app.listen(3000, () => {
  console.log('Server is running on http://localhost:3000');
});

Koa2 的优势

• 高性能：由于 Koa2 的轻量和无内置中间件设计，其性能表现优异。
• 灵活性：开发者可以自由选择和组合中间件，构建出适合自己需求的应用。
• 现代化：全面支持 async/await，代码更加简洁和易读。

Koa2 的局限性

• 学习曲线：对于新手来说，Koa2 的学习曲线可能比 Express 稍陡。
• 依赖第三方库：没有内置的路由、模板引擎等功能，需要依赖第三方库。

总结

Koa2 是一个现代、轻量且高效的 Node.js Web 框架，非常适合构建 API 服务和微服务。其模块化和中间件机制提供了极大的灵活性，使开发者可以根据需求自由组合功能。如果你喜欢现代 JavaScript 并且需要一个高性能的框架，Koa2 是一个非常不错的选择。

Koa 是一个由 Express 的原始团队开发的 Node.js Web 框架，旨在提供一个更小、更富表现力且更健壮的基础来编写 Web 应用和 API。以下是 Koa 的历史和发展过程：

Koa 框架的发展历史

1. 初创背景

Koa 的诞生背景与 Express 密切相关。Express 是最早的一批 Node.js Web 框架之一，由 TJ Holowaychuk 创建，并迅速成为最流行的 Node.js Web 框架之一。然而，随着 Node.js 和 JavaScript 生态系统的发展，Express 面临一些技术债务和设计局限性。

• Express 的局限性：
• 使用了回调函数来处理异步操作，容易导致“回调地狱”。
• 中间件的执行顺序较难管理。
• 错误处理机制不够完善。

2. Koa 的诞生

为了克服这些局限性，TJ Holowaychuk 和其他核心开发者开始设计一个新的框架，这个框架就是 Koa。Koa 的目标是提供一个更现代、更灵活的开发体验。

• 发布时间：Koa 的第一个版本于 2013 年发布。
• 主要设计理念：
• 轻量和模块化：Koa 本身非常轻量，没有内置的中间件，鼓励用户通过模块化的方式来构建应用。
• 现代 JavaScript：Koa 充分利用现代 JavaScript 特性，如 Generator 函数（在 Koa v1 中）和 async/await（在 Koa v2 中）。
• 中间件机制：采用洋葱模型的中间件机制，提供了灵活的请求响应流程控制。

3. Koa v1

Koa v1 使用了 ES6 的 Generator 函数来处理异步操作，这在当时是一个非常前沿的设计。Generator 函数提供了优雅的异步控制流，但它们的使用方式相对复杂，特别是对于新手开发者。

• Generator 中间件：Koa v1 的中间件是 Generator 函数，需要使用 yield next 来处理下一个中间件。
• 代码示例：

  const koa = require('koa');
  const app = new koa();

  app.use(function* (next) {
    this.body = 'Hello, Koa v1!';
    yield next;
  });

  app.listen(3000);

4. Koa v2

随着 ES7 async/await 的引入，JavaScript 的异步编程变得更加简单和直观。Koa 团队决定在 Koa v2 中采用 async/await 来取代 Generator 函数。

• 发布时间：Koa v2 于 2017 年正式发布。
• 主要改进：
• 使用 async/await 来处理异步操作，使代码更加简洁和易读。
• 保持轻量和模块化的设计理念。
• 代码示例：

  const Koa = require('koa');
  const app = new Koa();

  app.use(async (ctx, next) => {
    ctx.body = 'Hello, Koa v2!';
    await next();
  });

  app.listen(3000);

5. 发展和现状

Koa 在发布之后，迅速获得了社区的认可，特别是在需要高性能和灵活性的场景中，如 API 服务和微服务架构。由于其轻量和模块化的设计，Koa 被广泛用于构建各种类型的 Web 应用和服务。

• 社区和生态：Koa 有一个活跃的社区和丰富的中间件生态系统，开发者可以找到各种所需的中间件来扩展 Koa 的功能。
• 持续维护和更新：Koa 核心团队持续维护和更新框架，确保其与最新的 Node.js 和 JavaScript 规范保持兼容。

总结

Koa 是由 Express 的原始团队开发的，旨在提供一个更现代、更灵活的 Node.js Web 框架。通过采用 async/await 等现代 JavaScript 特性，Koa 提供了一个更优雅的异步编程模型。其轻量和模块化的设计使其成为构建高性能 Web 应用和 API 的理想选择。随着社区和生态系统的不断发展，Koa 仍然是一个受欢迎的 Node.js Web 框架。

CBOR，全称是简明二进制对象表示（Concise Binary Object Representation），是一种编码方式，常用于物联网（IoT）领域。它的设计目标是提供一种体积更小、更高效的二进制格式，类似于 JSON，但更适合资源受限的环境，如物联网设备。

CBOR 的特点

• 紧凑性：CBOR 的编码格式比 JSON 更紧凑，减少了数据传输的体积和存储空间。
• 高效性：由于其二进制格式，解析和生成 CBOR 数据通常比处理文本格式的 JSON 更高效。
• 自描述性：CBOR 编码的数据包含类型信息，解析时无需额外的模式（schema）。
• 广泛支持：大部分编程语言都有相应的 CBOR 编码和解码库，可以方便地处理 CBOR 数据。

CBOR 与 JSON 的比较

示例

以下是一个简单的 JSON 对象及其对应的 CBOR 编码表示：

JSON 示例：

{
  "name": "Alice",
  "age": 30,
  "is_student": false
}

CBOR 编码表示：

A3                                      # map(3)
  64                                    # text(4)
    6E616D65                            # "name"
  65                                    # text(5)
    416C696365                          # "Alice"
  63                                    # text(3)
    616765                              # "age"
  18 1E                                 # unsigned(30)
  6A                                    # text(10)
    69735F73747564656E74                # "is_student"
  F4                                    # false

CBOR 库

大部分编程语言都有相应的 CBOR 库，可以方便地进行编码和解码操作。以下是一些常见语言的 CBOR 库：

• Python: cbor2, cbor
• JavaScript: cbor
• Go: github.com/fxamacker/cbor
• Java: com.fasterxml.jackson.dataformat.cbor
• C/C++: libcbor

使用示例

Python 示例：

import cbor2

# 编码 JSON 对象为 CBOR
data = {
    "name": "Alice",
    "age": 30,
    "is_student": False
}
encoded = cbor2.dumps(data)
print(encoded)  # 输出 CBOR 二进制数据

# 解码 CBOR 为 JSON 对象
decoded = cbor2.loads(encoded)
print(decoded)  # 输出 {'name': 'Alice', 'age': 30, 'is_student': False}

结论

CBOR 提供了一种紧凑、高效的二进制编码格式，适用于资源受限的环境，如物联网设备。通过使用现有的 CBOR 库，可以轻松地在各种编程语言中进行 CBOR 数据的编码和解码操作，在提高数据传输效率的同时，保持了对各种复杂数据结构的支持。

Akka Streams 是基于 Reactive Streams 规范的流处理库，它提供了一种声明式的 API 来处理和传递数据流。Akka Streams 的核心概念包括：

1. Source：数据的起点，可以从中产生元素。
2. Flow：处理数据的步骤，可以对数据进行转换、过滤等操作。
3. Sink：数据的终点，接收从 Source 或 Flow 中传递过来的数据。
4. Materialization：流的实际执行过程，这个过程会产生一个运行时值（如 Future）。

通过组合 Source、Flow 和 Sink，可以构建出复杂的数据流处理逻辑。

基本示例

我们通过一个简单的例子来说明 Akka Streams 的基本概念。

import akka.actor.ActorSystem
import akka.stream.scaladsl.{Flow, Sink, Source}
import akka.stream.{ActorMaterializer, Materializer}
import scala.concurrent.Future
import scala.util.{Failure, Success}

object AkkaStreamsExample extends App {
  implicit val system: ActorSystem = ActorSystem("example-system")
  implicit val materializer: Materializer = Materializer(system)
  import system.dispatcher  // 用于处理 Future 的回调

  // 创建一个 Source，从1到10的整数序列
  val source: Source[Int, NotUsed] = Source(1 to 10)

  // 创建一个 Flow，对每个元素乘以2
  val flow: Flow[Int, Int, NotUsed] = Flow[Int].map(_ * 2)

  // 创建一个 Sink，打印每个接收到的元素
  val sink: Sink[Int, Future[Done]] = Sink.foreach[Int](println)

  // 将 Source、Flow 和 Sink 连接起来，形成一个流
  val runnableGraph: RunnableGraph[Future[Done]] = source.via(flow).toMat(sink)(Keep.right)

  // 运行流
  val result: Future[Done] = runnableGraph.run()

  // 处理流完成后的结果
  result.onComplete {
    case Success(_) =>
      println("Stream completed successfully")
      system.terminate()
    case Failure(e) =>
      println(s"Stream failed with e") system.terminate() }
最后，我们监听 result 的完成情况，打印结果并终止 ActorSystem。