使用Scala构建微服务：Akka持久化与HTTP服务实现

立即解锁

发布时间: 2025-08-19 00:05:42 阅读量: 6 订阅数: 16

Scala 2.13编程实战与进阶

### 使用 Scala 构建微服务：Akka 持久化与 HTTP 服务实现在软件开发中，构建微服务时采用事件溯源（Event Sourcing）是一种强大的模式，它允许我们将系统状态的变化以事件的形式存储，从而实现系统状态的恢复和分析。本文将详细介绍如何使用 Scala 和 Akka 框架来构建一个基于事件溯源的微服务，并通过 Akka-HTTP 提供 API 接口。 #### 事件溯源的优势与挑战事件溯源模式具有显著的优势，同时也存在一些挑战： - **优势**： - **状态恢复**：能够将系统状态恢复到过去的任何特定时刻。 - **灵活的状态表示**：事件可以以不同的方式组合，从而构建不同的状态表示。 - **历史数据分析**：结合状态恢复的优势，我们可以以事件创建时未知的方式分析过去的数据。 - **挑战**： - **状态重建**：状态在从事件中重建之前并不存在，可能需要编写特殊代码来分析事件，构建状态表示需要一定的努力。 - **领域模型膨胀**：在复杂项目中，实现新的用例通常需要引入新的命令和事件。 - **模型变更**：随着项目的发展，现有用例的变更可能意味着现有事件结构的变更，由于事件日志是追加式的，这些变更需要在代码中进行。 - **事件数量增长**：在活跃使用的系统中，事件数量可能会迅速增长，影响构建状态表示所需的时间，通常使用快照（Snapshotting）来解决这个问题。 #### 配置 Akka 持久化 Akka 持久化允许我们存储和重放发送给持久化参与者（PersistentActor）的消息，从而实现事件溯源方法。在开始实现参与者之前，我们需要进行项目配置。 ##### 数据库选择与表结构我们将使用 H2 关系型数据库，并使用 Flyway 来创建数据库结构。以下是存储事件和快照的表结构： ```sql -- 存储事件的表 CREATE TABLE IF NOT EXISTS PUBLIC."journal" ( "ordering" BIGINT AUTO_INCREMENT, "persistence_id" VARCHAR(255) NOT NULL, "sequence_number" BIGINT NOT NULL, "deleted" BOOLEAN DEFAULT FALSE, "tags" VARCHAR(255) DEFAULT NULL, "message" BYTEA NOT NULL, PRIMARY KEY("persistence_id", "sequence_number") ); -- 存储快照的表 CREATE TABLE IF NOT EXISTS PUBLIC."snapshot" ( "persistence_id" VARCHAR(255) NOT NULL, "sequence_number" BIGINT NOT NULL, "created" BIGINT NOT NULL, "snapshot" BYTEA NOT NULL, PRIMARY KEY("persistence_id", "sequence_number") ); ``` - `persistence_id`：特定持久化参与者的 ID，在整个参与者系统中必须唯一。 - `tags`：分配给事件的标签，便于构建视图。 - `message`：序列化形式的事件。 ##### 添加依赖在 `build.sbt` 中添加以下依赖： ```scala "com.typesafe.akka" %% "akka-persistence" % akkaVersion, "com.github.dnvriend" %% "akka-persistence-jdbc" % akkaPersistenceVersion, "com.scalapenos" %% "stamina-json" % staminaVersion, "com.h2database" % "h2" % h2Version, "org.flywaydb" % "flyway-core" % flywayVersion, ``` - `akka-persistence`：Akka 持久化的核心依赖。 - `akka-persistence-jdbc`：H2 数据库的 JDBC 存储实现。 - `stamina-json`：支持模式迁移，允许将数据库中旧格式存储的事件转换为代码中使用的新格式。 - `h2`：H2 数据库。 - `flyway-core`：用于设置数据库结构。 ##### 配置序列化在 `application.conf` 中配置 Akka 持久化的序列化： ```scala akka.actor { serializers.serializer = "ch14.EventSerializer" serialization-bindings { "stamina.Persistable" = serializer } } ``` 以下是 `EventSerializer` 的实现： ```scala class EventSerializer extends stamina.StaminaAkkaSerializer(v1createdPersister, v1deletedPersister, v1purchasedPersister, v1restockedPersister, v1inventoryPersister) ``` 在 `PersistenceSupport` 对象中定义事件的持久化器： ```scala import stamina.json._ object PersistenceSupport extends JsonSupport { val v1createdPersister = persister[ArticleCreated]("article-created") val v1deletedPersister = persister[ArticleDeleted]("article-deleted") val v1purchasedPersister = persister[ArticlesPurchased]("articles-purchased") val v1restockedPersister = persister[ArticlesRestocked]("articles-restocked") val v1inventoryPersister = persister[Inventory]("inventory") } ``` 在 `JsonSupport` 特质中提供隐式的 `RootJsonFormat`： ```scala import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport import spray.json.{DefaultJsonProtocol, RootJsonFormat} import DefaultJsonProtocol._ trait JsonSupport extends SprayJsonSupport { implicit val invJF: RootJsonFormat[Inventory] = jsonFormat1(Inventory) implicit val createArticleJF = jsonFormat2(CreateArticle) implicit val deleteArticleJF = jsonFormat1(DeleteArticle) implicit val purchaseJF = jsonFormat1(PurchaseArticles) implicit val restockJF = jsonFormat1(RestockArticles) implicit val createdJF = jsonFormat2(ArticleCreated) implicit val deletedJF = jsonFormat1(ArticleDeleted) implicit val pJF = jsonFormat1(ArticlesPurchased) implicit val reJF = jsonFormat1(ArticlesRestocked) } ``` ####