[翻译]akka in action之akka-stream（2 流式HTTP）

「爱情、让人受尽委屈。」 2021-12-04 02:15 307阅读 0赞

## 2 流式HTTP ##

日志流处理器（log-stream processor ）将以HTTP服务运行。让我们来看看这意味着什么。Akka-http 使用 akka-stream，所以从基于文件的APP到HTTP服务并不需要很多粘合代码。Akka-http 是一个非常好的例子，一个包含akka-stream的库。

首先，我们要在工程中添加更多的依赖：

这一次我们将构建LogsApp，可以从某些存储器读写流日志。在本例中，为了保持简单，我们将流直接写入文件。

有相当多的响应式的基于流的客户端库可用。将示例连接到某种其他类型的 (数据库) 存储, 作为读者的练习。

## 2.1 HTTP上接收流 ##

我们允许服务的客户端使用HTTP POST流化日志事件数据。数据将存储到服务器的某个文件中。POST对URL /logs/\[log\_id\]将在logs目录下创建名为\[log\_id\]的文件。稍后，当对/logs/\[log-id\]执行GET命令时，我们将从这个文件获取流。此处省略了对HTTP服务器的设置。

HTTP路由在LogsApi类中定义，如下所示。logsDir 指向了日志存储的目录。logFile 方法仅返回文件的ID。EventMarshalling特质混入了支持JSON编组。你会注意到ExecutionContext 和 ActorMaterializer 在隐式范围内，运行Flow它们是必需的。

我们将使用上一节中的 BidiFlow, 因为它已经定义了从日志文件到Event JSON 的协议。下面的代码展示了将要使用的Flow、Sink和Source，当执行HTTP GET时，我们将返回它。

![aHR0cHM6Ly9waWM0LnpoaW1nLmNvbS84MC92Mi01MDkzYjUzMjA5YzY3ZDgxOWFmY2U5ZDEyMmE4N2RjYl9oZC5qcGc][]

在POST中使用Flow和Sink

在本例中, 事件保持不变;所有日志行都将转换为 JSON 事件。将与 BidiFlow 连接的流改造为基于查询参数过滤事件的方法留给读者。

> **响应前完全读取实体Source**  
> 从databytessource完全读取所有的数据是很重要的。如果在从源读取所有数据之前进行响应, 例如, 使用 HTTP 永久连接的客户端可能会确定 TCP 套接字仍然适合用于下一个请求；这可能会导致尚未再次读取未完全读取的源，连接就结束了。  
> HTTP客户端通常认为，请求将被完全处理，因此在请求完全被处理之前，它不会尝试读取响应。即使你不使用永久连接，也最好完全处理请求。  
> 这通常是阻塞型 HTTP 客户端的一个问题, 它在写入整个请求之前不会开始读取响应。  
> 这并不意味着请求/响应周期是同步处理的。在本节的示例中, 响应是在处理请求后异步返回的。

HTTP POST 在postRoute 方法中处理，如下所示。因为akka-http建立在akka-stream之上，在HTTP上接收流相当简单。HTTP请求体有个dataBytes Source，我们可以从中读取数据。

![aHR0cHM6Ly9waWMxLnpoaW1nLmNvbS84MC92Mi1mOWJkYWVjM2ZlYjQ5MDc0YzFiOTVmNDM2NDc2OWEzNF9oZC5qcGc][]

处理POST

run方法返回了一个Future\[IOResult\]，所以我们使用onComplete指令，它最终将Future的结果传递给内部路由，此处将处理Success和Failure情况。响应将用complete指令返回。

在下一节，我们将看看一个HTTP GET请求的响应，该响应会以JSON格式流化日志文件并返回给客户端。

## 2.2 HTTP上用流响应 ##

使用HTTP GET，客户端能够取回日志事件流。实现路由如下：

![aHR0cHM6Ly9waWMzLnpoaW1nLmNvbS84MC92Mi05Njc4NDAxODhhNGJhYjkwZDRiMTQ0NmJhYmIzMDZkMl9oZC5qcGc][]

处理GET

> 标识符中的引号  
> Akka-http 尽可能贴近HTTP规范，这也体现在对HTTP头，内容类型和HTTP规范其他元素标识符的命名上。创建标识符时，通常不允许而在HTTP规范中常用的字符，例如破折号和斜线，在Scala中可以使用引号创建。

HttpEntity有一个含有一个ContentType和一个Source的apply方法。从文件流化数据的Source传递给该方法，使用complete指令完成响应。在POST例子中，我们简单的假设以期望的日志格式用文本发送。在GET例子中，我们用JSON格式返回数据。

现在, 我们已经有了最简单的流化GET和POST 示例, 让我们来看一下如何使用akka-http 进行内容协商, 这将使客户端能够以 JSON 或日志格式GET和POST 数据。

## 2.3 用于内容类型和协商的自定义 marshallers 和 unmarshallers ##

如果有多个媒体类型可用，则Accept头部允许HTTP客户端指定它想要GET的格式。HTTP客户端可以设置Content-Type 头部类指定POST中实体的格式。我们将在这一节中对这两种情况进行处理, 使得可以以 JSON 或日志格式交换 POST 和GET

数据, 类似于 BidiEventFilter 示例。

幸运的是，akka-http提供了自定义编组和解组，负责内容协商，这意味着我们更少的工作。让我们开始于处理POST里的Content-Type头部。

**在自定义解组中处理Content-Type**

Akka-http为许多预定义类型，比如来自实体、字节数组、字符串等等数据，提供了解组。也可以自定义解组。本例，我们仅支持两种内容类型：用于表示日志格式的text/plain和用于表示JSON格式日志事件的 application/json。基于Content-Type，entity.dataBytes source或者用分隔线或者用JSON帧化，并像往常一样处理。

Unmarshaller特质仅需要实现一个方法。

![aHR0cHM6Ly9waWMzLnpoaW1nLmNvbS84MC92Mi01YTM2OGEzNmQyYTBiZDU0NzRlMTljYjZlOGQzMTkzYV9oZC5qcGc][]

在EventUnmarshaller里处理Content-Type

create方法创建了一个匿名的Unmarshaller 实例。apply方法首先创建了一个Flow用于处理传入的数据，并通过via方法与dataBytesSource组合成一个新的Source。

这个Unmarshaller 必须在隐式范围内，以便实体指令可用于提取Source\[Event, \_\]，可以在ContentNeg-Logs-Api 类中找到。

![aHR0cHM6Ly9waWMyLnpoaW1nLmNvbS84MC92Mi1hMzM5NWVkMTIyNTQ4MGYwZjAxZWQ4NDdjZDhhYTBkNV9oZC5qcGc][]

在POST中使用EventUnmarshaller

尝试 aia.stream.ContentNegLogsApp 是留给读者的练习。请务必指定Content-Type, 例如使用 httpie。如下是使用httpie的POST例子，使用了Content-Type 头部：

下面，我们将看看使用自定义的编组，为内容协商处理Accept头部。

**使用自定义编组进行内容协商**

我们将写一个自定义的Marshaller在响应中来支持text/plain和application/json内容类型。Accept头部能用于指定响应可接受的媒体类型。如下所示（使用httpie）：

![aHR0cHM6Ly9waWMyLnpoaW1nLmNvbS84MC92Mi1kMzY3MDE5OTY0NjdiNmQ2ZDI1NWI5MTczMTA2NzM0ZF9oZC5qcGc][]

httpie的GET例子，使用Accept头部

客户端可以表示它只接受特定的Content-type, 或者它具有特定的首选项。确定应用哪个Content-type进行响应的逻辑是在Akka中实现的。我们所要做的就是创建一个支持一组内容类型的Marshaller。

LogEntityMarshaller 对象创建了ToEntityMarshaller。

![aHR0cHM6Ly9waWMxLnpoaW1nLmNvbS84MC92Mi0wOTIxYzE0Y2I2MDBhMzY5ZTI1MmMxNjhiYTIyMTY1NF9oZC5qcGc][]

提供编组用于内容协商

Marshaller.withFixedContentType 是一个方便的方法用于根据指定的Content-Type创建Marshaller。它带有一个函数，A => B，本例中是Source\[ByteString, Any\] => HttpEntity。src提供了JSON日志文件的字节，它转化为一个HttpEntity。

LogJson.jsonToLogFlow 方法使用了先前我们使用过的相同技巧，将一个 Flow\[Event\]加入BidiFlow ，这次是JSON到日志格式。

这个Marshaller必须放在隐式范围内，以便HTTP GET路由中使用。

![aHR0cHM6Ly9waWMzLnpoaW1nLmNvbS84MC92Mi1iOTcyN2FhYzEyZDhmYjI1NGVlM2M3ZjE2NzczMmEwYV9oZC5qcGc][]

GET路由

Marshal(src).toResponseFor(req)采用日志文件Source，并根据请求（包括Accept 头部）创建响应，其中以LogEntityMarshaller设置内容协商。

这就结束了使用Content-Type 头部支持两种格式和使用Accept头部进行内容协商的示例。

LogsApi 和 ContentNegLogsApp 读写事件不变。我们可以在请求时对它们的状态进行筛选, 但将事件根据状态进行拆分 (OK, warning, error, critical) 并将这些事件存储在不同的文件中是更有意义的, 例如, 可以检索所有错误, 而无需每次都要过滤。下一节，我们将看一看如何在akka-stream中扇入扇出。我们将根据状态将事件存储在服务器上不同的文件中，但我们也可以检索部分状态，像所有非OK状态的事件。

> JSON流化支持  
> 本例支持文本日志格式和JSON格式的日志事件。当你只想支持JSON，这有个更简单的选项。akka.http.scaladsl.common 包下的EntityStreamingSupport 对象通过EntityStreamingSupport.json提供了一个JsonEntityStreamingSupport，当放入隐式范围时，可以直接使用complete(events)直接完成HTTP请求。也可以从entity(asSourceOf\[Event\])直接获取Source\[Event, NotUsed\]。

[aHR0cHM6Ly9waWM0LnpoaW1nLmNvbS84MC92Mi01MDkzYjUzMjA5YzY3ZDgxOWFmY2U5ZDEyMmE4N2RjYl9oZC5qcGc]: /images/20211203/845d8e689efb4a4a9fdfcd0d289fde45.png
[aHR0cHM6Ly9waWMxLnpoaW1nLmNvbS84MC92Mi1mOWJkYWVjM2ZlYjQ5MDc0YzFiOTVmNDM2NDc2OWEzNF9oZC5qcGc]: /images/20211203/f8fdce4e6583426088a2b8652f2c66b1.png
[aHR0cHM6Ly9waWMzLnpoaW1nLmNvbS84MC92Mi05Njc4NDAxODhhNGJhYjkwZDRiMTQ0NmJhYmIzMDZkMl9oZC5qcGc]: /images/20211203/fb1a9eef416b41c9b87b526f6694463e.png
[aHR0cHM6Ly9waWMzLnpoaW1nLmNvbS84MC92Mi01YTM2OGEzNmQyYTBiZDU0NzRlMTljYjZlOGQzMTkzYV9oZC5qcGc]: /images/20211203/7f40f150b2864ac49b9e03e734156056.png
[aHR0cHM6Ly9waWMyLnpoaW1nLmNvbS84MC92Mi1hMzM5NWVkMTIyNTQ4MGYwZjAxZWQ4NDdjZDhhYTBkNV9oZC5qcGc]: /images/20211203/2f213aa284e64f7c8e35810bbf4dff56.png
[aHR0cHM6Ly9waWMyLnpoaW1nLmNvbS84MC92Mi1kMzY3MDE5OTY0NjdiNmQ2ZDI1NWI5MTczMTA2NzM0ZF9oZC5qcGc]: /images/20211203/226ac7f4a2d44946b63df1d89767aa2e.png
[aHR0cHM6Ly9waWMxLnpoaW1nLmNvbS84MC92Mi0wOTIxYzE0Y2I2MDBhMzY5ZTI1MmMxNjhiYTIyMTY1NF9oZC5qcGc]: /images/20211203/a9e88e029caa45358154c781d5b945d8.png
[aHR0cHM6Ly9waWMzLnpoaW1nLmNvbS84MC92Mi1iOTcyN2FhYzEyZDhmYjI1NGVlM2M3ZjE2NzczMmEwYV9oZC5qcGc]: /images/20211203/a4276356f6ef4be1a8bcbc142ebb8c23.png