前言

上篇主要是讲解理论知识和项目架构要点，这篇将集中在微服务中使用spring Boot、Spring Cloud和Project Reactor实现事件溯源的原始主题。文章中也会介绍项目实现一些技术细节，项目Git下载地址：https://github.com/kbastani/spring-cloud-event-sourcing-example，项目我自己已经运行了一遍，非常适合学习使用。下面是原文翻译内容：

Project Reactor

Project Reactor 是一个开源基于JVM实现Reactive流规范的开发框架，是Spring生态系统一个成员，在微服务中，经常在一个上下文下需要和其他微服务交互操作，由于微服务架构天然属性是最终一致性，而最终一致性并不保证数据的安全性。它提供我们一个使用异步非堵塞方式进行通讯的方式，这里正是使用Reactor目的所在。

很少情况下，领域模型状态会被跨微服务共享，但是如果在微服务之间需要共享状态怎么办？或者说多个微服务需要访问同一个数据库数据表怎么办？在微服务中ES只保存有序事件的日志，使用事件流取代领域模型在数据库中存储，我们可以存储有序事件流来代表对象的状态，这样，意味著我们就不再使用基于HTTP的RESTful进行微服务之间同步通讯，这些同步会造成堵塞和性能延迟。

Reactor为核心的事件溯源

在网上商店的微服务之一是购物车服务（Shopping Cart Service），已验证的用户从商店Web应用程序的用户界面浏览产品目录。用户可以添加和删除他们的购物车的商品条目，以及清除他们的购物车或结帐。

一个用户的购物车为如何实现事件追溯工作描绘了一个简单的样例。我们以网上商店中购物车服务为案例，展示Reactor + ES是如何实现的：

购物车服务Shopping Cart Service是一个MySQL数据库拥有者，有一个数据表称为cart_event。这个表包含用户操作动作产生的有序事件日志，用户操作就是反复将商品加入购物车或去除等各种购物车管理操作。

1. Example 1. CartEventType.java
2. // These will be the events that are stored in the event log for a cart
3. public enum CartEventType {
4. ADD_ITEM,
5. REMOVE_ITEM,
6. CLEAR_CART,
7. CHECKOUT
8. }

CartEventType是枚举类型，已经列出了4种不同的事件类型。这些事件类型中的每一个都代表用户在购物车上执行的动作。根据ES，这些购物车事件可以影响用户的购物车的最终状态结果。当用户添加或删除一个商品条码到他们的购物车时，一个动作产生一个事件，会对购物车中进行递增或递减一行条目。当这些事件使用同样顺序进行回放时，同样一系列的条目会被重新创建或删除：

1. id  created_at  last_modified   cart_event_type product_id  quantity    user_id
2. 1   1460990971645   1460990971645   0   SKU-12464   2   0
3. 2   1460992816398   1460992816398   1   SKU-12464   1   0
4. 3   1460992826474   1460992826474   0   SKU-12464   2   0
5. 4   1460992832872   1460992832872   0   SKU-12464   2   0
6. 5   1460992836027   1460992836027   1   SKU-12464   5   0

我们看到每行都有一个唯一时间戳来确保严格顺序，使用整数来代表4个购物车事件类型，product_id 和数据quantity都是每次加入购物车的商品条码信息。

这一结果显示在上面的截图，在这里，我们看到一个用户的购物车，生成作为一个总结果集的对象。

选择事件存储库

当选择事件追溯的适当存储选项时，有很多可用的选项。今天几乎所有的数据库都能提供数据流查询功能的工作，然而，有一些流行的开源项目，在这方面有着突出的优点。现在Event Sourcing标准的存储库是 Apache Kafka，微服务之间共享状态是通过共享Kafka的事件日志实现的，这是一个未来博客的主题。在这个例子中我们将使用MySQL，这是实现事件追溯一个在线购物车的不错选择。

您的事件存储技术的选择将永远取决于写入的数量和您的数据库的吞吐量。像Apache Kafka设计了精确的使用情况，却要求我们承担一些额外的工作责任去在生产中扩展，包括运行Apache ZooKeeper集群。

生成聚合

下面我们回到购物车，购物车微服务提供一个REST API方法接受来自Web端的事件。Web端发出事件的控制器 ShoppingCartControllerV1.Java

1. Example 2. ShoppingCartControllerV1.java
2. @RequestMapping(path = "/events", method = RequestMethod.POST)
3. public ResponseEntity addCartEvent(@RequestBody CartEvent cartEvent) throws Exception {
4. return Optional.ofNullable(shoppingCartService.addCartEvent(cartEvent))
5. .map(event -> new ResponseEntity(HttpStatus.NO_CONTENT))
6. .orElseThrow(() -> new Exception("Could not find shopping cart"));
7. }

在上面的代码示例，我们定义了一个用于收集来自客户端新的CartEvent对象的控制器方法。这种方法的目的是在向事件日志追加事件。当客户端调用REST API检索用户的购物车，它将产生一个购物车聚合，使用Reactive流合并了所有购物车事件流。

下面在ShoppingCartServiceV1.java中使用Reactor产生购物车事件流：

1. Example 3. ShoppingCartServiceV1.java
2. public ShoppingCart aggregateCartEvents(User user, Catalog catalog) throws Exception {
3. // Create a reactive streams publisher by streaming ordered events from the database
4. Flux<CartEvent> cartEvents =
5. Flux.fromStream(cartEventRepository.getCartEventStreamByUser(user.getId()));
6. // Aggregate the current state of the shopping cart until arriving at a terminal state in the stream
7. ShoppingCart shoppingCart = cartEvents
8. .takeWhile(cartEvent -> !ShoppingCart.isTerminal(cartEvent.getCartEventType()))
9. .reduceWith(() -> new ShoppingCart(catalog), ShoppingCart::incorporate)
10. .get();
11. // Generate the list of line items in the cart from the aggregate
12. shoppingCart.getLineItems();
13. return shoppingCart;
14. }

在上面的代码示例中，我们可以看到三个步骤来生成购物车对象，然后返回到客户端。第一步是从事件存储的数据源中创建一个Reactive流。一旦流建立，我们可以从事件流中产生我们的聚合。这些事件流不断改变购物车状态直至到最终状态，然后就可以将最终购物返回给用户客户端。

为了减少reactive流的聚合，我们使用了一个称为incorporate方法，这个方法是接受CartEvent对象，而CartEvent对象是用来改变购物车状态的。

1. Example 4. ShoppingCart.java
2. public ShoppingCart incorporate(CartEvent cartEvent) {
3. // Remember that thing about safety properties in microservices?
4. Flux<CartEventType> validCartEventTypes =
5. Flux.fromStream(Stream.of(CartEventType.ADD_ITEM,
6. CartEventType.REMOVE_ITEM));
7. // The CartEvent's type must be either ADD_ITEM or REMOVE_ITEM
8. if (validCartEventTypes.exists(cartEventType ->
9. cartEvent.getCartEventType().equals(cartEventType)).get()) {
10. // Update the aggregate view of each line item's quantity from the event type
11. productMap.put(cartEvent.getProductId(),
12. productMap.getOrDefault(cartEvent.getProductId(), 0) +
13. (cartEvent.getQuantity() * (cartEvent.getCartEventType()
14. .equals(CartEventType.ADD_ITEM) ? 1 : -1)));
15. }
16. // Return the updated state of the aggregate to the reactive stream's reduce method
17. return this;
18. }

在上面代码中我们看到ShoppingCart的incorporate方法实现，我们接受一个CartEvent对象然后做最重要的一件事：确保事件类型是正确的。这是在微服务需要自由与他们的单元测试，在最终一致的架构，以确保状态突变将确保数据的正确性。在本例中，我们确保事件类型是 ADD_ITEM 或 REMOVE_ITEM。

下一步是更新购物车中每个条目的聚合视图，通过映射相应的事件类型到商品条目的数量递增或递减。最后我们返回这样一个带有最终可变状态的购物车给客户端。

Docker Compose演示

示例项目使用Docker Compose构建和运行，每个微服务的容器镜像都作为Maven编译过程的一部分。

入门

首先，访问该示例项目的GitHub库：

1. https://github.com/kbastani/spring-cloud-event-sourcing-example

克隆或复制项目并将存储库下载到您的机器上。下载后，您将需要使用Maven和Docker来编译和构建本地镜像。

下载Docker

首先，如果你还没有下载Docker。可遵循这里的指示https://www.docker.com/docker-toolbox，把Docker toolbox在你的开发机器上运行。

在你安装了Docker toolbox以后，运行下面的命令来初始化这个示例应用程序的一个新的VirtualBox虚拟机。

1. $ docker-machine create env event-source-demo --driver virtualbox --virtualbox-memory "11000" --virtualbox-disk-size "100000"
2. $ eval "$(docker-machine env event-source-demo)"

环境要求

能够运行实例程序，需要在你的开发机上安装下面的软件：

• Maven 3
• Java 8
• Docker
• Docker Compose

构建项目

通过命令行方式来构建当前项目，在项目的根目录中运行如下的命令：

1. $ sh run.sh

该项目将下载所有所需的依赖关系，并编译每一个项目的组件。每个服务都将被构建，然后一个Maven Docker插件会自动构建每个镜像到你的本地Docker注册表里。在根目录运行命令之前，一定要保证Docker正常运行，只有这样，你的sh run.sh命令才会构建成功。

Docker Compose启动集群

现在，每个镜像都已经搭建成功，我们可以使用Docker Compose快速启动集群。run.sh脚本将建立每个项目和Docker容器，并使用Docker Compose开启每个服务。值得注意的是，集群需要首先启动的服务是配置服务和发现服务。其余的服务在开始启动，并最终开始相互沟通。

我强烈建议你运行此示例的机器上至少16GB的系统内存。

一旦启动序列完成后，你可以浏览到Eureka主机和看到发现服务里注册的所有服务。复制并粘贴以下命令到终端，Docker可以使用$DOCKER_HOST环境变量进行访问。

1. $ open $(echo \"$(echo $DOCKER_HOST)\"|
2. \sed 's/tcp:\/\//http:\/\//g'|
3. \sed 's/[0-9]\{4,\}/8761/g'|
4. \sed 's/\"//g')

如果Eureka正确的启动，浏览器将会启动并打开Eureka服务的仪表盘，如下图所示：

当所有的应用程序在Eureka上已经完成启动和注册，你可以使用以下命令访问网上商店的Web应用。

1. $ open $(echo \"$(echo $DOCKER_HOST)\"|
2. \sed 's/tcp:\/\//http:\/\//g'|
3. \sed 's/[0-9]\{4,\}/8787/g'|
4. \sed 's/\"//g')

应用程序启动可能需要一些时间，所以请确保你每隔几分钟刷新用户界面，直到看见产品目录。

要开始向购物车添加产品，您需要登录默认用户。单击Login按钮，你将被重定向到身份验证网关页面，在此页面你可以使用用户的默认凭据和密码进行登录。

登录成功后，你将被重定向到需要身份验证才能呈现的主页面上，并可以开始管理你的购物车中的项目。

总结

在这篇文章中，我们很难看到在微服务架构中的高可用性和数据一致性的挑战。我们期待一个完整的网上商店原生云应用程序，能作为一系列微服务的集合，使用事件追溯保持一致的世界观，同时还保证高可用性。

在接下来的博客，我将继续探索如何在事件追溯中使用Spring Cloud Stream和用Apache Kafka对事件流处理。