NestJS小技巧35-最近，我尝试在微服务架构中用SAGA模式

原文链接

昨天，我尝试做一个简单微服务应用，这个应用是一个小项目，是我为自己构建的为了发布到云平台。因为我从来没有发布微服务到云平台，当服务发布完以后，我不知道每个服务之间怎么通信。我上一个项目已经发布到云平台了是一个简单的博客项目，而且这也不是微服务，我使用云平台就像我使用VPS一样。所以我决定学习云技术和云基础设施相关的知识，也学习Kubernetes，最后我用微服务模式构建了一个简单的在线仓库。

当我决定做这些事情的时候，我感觉我很懒。实际上，当我构建项目且这个项目不是我公司的工作或者不是我的重要项目时，我是一个懒人。我想偷懒而且我觉得我可以玩玩Duolingo（学习英语的平台）来学英语和基本的日语是更好的一天，甚至比用Golang和一些很好的技术来构建这些还要重要。我决定使用NestJs，因为它拥有出色的代码生成功能。目前，我对Saga模式的实现仍然在HTTP层的最上方。也许在下一个迭代中，我会迁移到消息传递或基于事件的实现。也可能会在文章中更新有关基于事件的内容。

问题

当我第一次知道微服务（大概2018年）我认为微服务就是一个拆分单体应用的一个技术。就像当我有一个电子商务应用时，我可以将其拆分成不同的服务，比如用户服务、产品服务等。并且我认为只有这种方法才被称为微服务。但是当我深入学习其他的微服务方法，像拆分数据库，因此每个服务都有单独的数据库。然后我尝试这个方法在数据库中来做电商来处理各种交易我发现了问题。“我怎么才能在我的服务中实现使用不同数据库的时的交易，因为这些方法不能实现ACID”。然后我认识到了Saga模式

Saga

Saga设计模式是一种数据访问分布式事务的微服务时保持同步方式。因为ACID不被支持分布式事务。saga是一系列事务，当这个事务成功，它会发送事件/消息到事务的下一步，但是如果这个事务失败Saga也会发送失败的事件/消息给每个服务。

在单体应用程序或使用单一数据库的微服务中，我们可以使用两阶段提交来保持事务中的数据一致性。两阶段提交每个数据库事件像创建，读，更新，和删除在一个事务中应该是在同一个时间运行提交或者回滚。因此数据会在同一个时间内插入数据库或者不插入。但是在多数据库两阶段提交不是一个好的选择。

在Saga模式中，有两种不同的方法：

1. Choreography（协作）
2. Orchestration（编排）

Choreography

在Choreography模式下，没有核心服务来处理所有的事物流。所有的服务来处理他们自己的工作进程有他们的消费事件和发布事件

在我尝试学习Saga模式的概念以后。我决定使用Orchestration模式因为它比Choreography更加的简单。以我个人的见解，OrchestrationSaga是更加容易构建和调试。因为它有一个编排器来管理业务逻辑，而不像Choreography模式那样，其中每个服务都是独立的，都有自己的业务逻辑，需要确保是否继续下一个流程。此外，在Choreography模式中，要了解应用程序的流程会更加困难，因为您需要逐个阅读所有具有关联的服务中的流程。

所以这是我的应用的Saga示意图

流程：

1. 首先，下单服务将扮演编排器。
2. 在下单服务中基于cart_id创建下单,如果成功了则基于cart_id继续删除用户的购物车
3. 下单服务将会调用购物车服务来删除用户的购物车，在删除购物车之前，我使用Redis保存购物车到缓存中。如果成功了，接着下一步
4. 下单服务将会调用产品服务来减少产品的数量。但是在我减少产品数量之前，我也把产品数量放到redis里了。
5. 如果所有的流程都成功了，执行提交。
6. 在购物车服务和产品服务执行提交将会移除Redis中的缓存

所以如果异常发生了，我应该怎么回滚呢

流程：

1. 首先，下单服务将扮演编排器。
2. 在下单服务中基于cart_id创建下单,如果成功了则基于cart_id继续删除用户的购物车。
3. 下单服务将会调用购物车服务来删除用户的购物车，并且异常发生了，之后。。
4. 回滚下单
5. 回滚购物车
6. 回滚产品

我也使用Redis来在出现故障时保留我的数据。因此，在购物车服务中，在删除用户的购物车之前，我会查找购物车并将其保存到Redis中，然后再删除购物车，对于产品也是如此。所以在订单服务中，我只是对本地订单服务事务、购物车服务和产品服务进行提交。在购物车服务和产品服务中，提交函数将从Redis中删除数据。而产品和购物车服务的回滚函数将以前保存在Redis中的数据插入到MySQL数据库中，并删除Redis中的数据。

在代码中实现

在下单服务添加新的下单

下面是我写的代码在订单服务中创建一个订单

async CreateOrderByUserCart(
    auth: string,
    cartIds: string[],
  ): Promise<string> {
    const userCarts = await this.cartService.GetUserCartByCartIds(
      auth,
      cartIds,
    );

    if (userCarts.data.length === 0) {
      return 'cart_is_empty';
    }

    const productVariantIds = userCarts.data.map(
      (data) => data.product_variant_id,
    );

    const mutex = this.mutex.NewMutex();

    const result = await mutex.runExclusive(async () => {
      // 首先添加一个定的
      // 取得连接并创建query runner
      const connection = getConnection();
      const queryRunner = connection.createQueryRunner();

      await queryRunner.connect();

      await queryRunner.startTransaction();

      try {
        const order = new Orders();
        order.user_id = userCarts.data[0].user_id;
        order.total_quantity = userCarts.data.reduce(
          (prevVal: number, curVal: CartsFromExternal) =>
            (prevVal += curVal.quantity),
          0,
        );
        order.total_price = userCarts.data.reduce(
          (prevVal: number, curVal: CartsFromExternal) =>
            (prevVal += curVal.price),
          0,
        );
        order.order_status = OrderStatus.PENDING;
        const newOrder = await queryRunner.manager.save(order);

        const orderDetails: OrdersDetail[] = [];

        for (const cart of userCarts.data) {
          const orderDetail = new OrdersDetail();
          orderDetail.price = cart.price;
          orderDetail.quantity = cart.quantity;
          orderDetail.product_variant_id = cart.product_variant_id;
          orderDetail.orders = newOrder;
          orderDetails.push(orderDetail);
        }
        order.order_detail = orderDetails;

        await queryRunner.manager.save(orderDetails);

        // 在保存完下单数据之后删除用的的购物车
        if (cartIds.length > 1) {
          await this.cartService.DeleteBulkUserCart(auth, cartIds);
        } else {
          await this.cartService.DeleteUserCart(auth, cartIds[0]);
        }
        // 在保存完下单数据后更新产品的数量
        await this.productService.PatchProductVariantStockByVariantId(
          auth,
          userCarts.data,
        );
        await this.productService.CommitPatchProductVariantStockByVariantId(
          auth,
        );
        await this.cartService.CommitDeleteUserCart(auth);
        await queryRunner.commitTransaction();
        console.log('order - success');
        return 'success make order';
      } catch (e) {
        console.log('order - failed');
        console.log(e.message);
        await queryRunner.rollbackTransaction();
        await this.cartService.RollbackUserCart(auth);
        await this.productService.RollbackPatchProductVariantStockByVariantId(
          auth,
        );
        return 'failed make order';
      } finally {
        await queryRunner.release();
      }
    });

    return result;
  }

就像您看到的这样。我的服务有个本地事务用来处理下单事务，如果成功事务将被提交，如果失败将会执行回滚。在61-65行，我调用了购物车服务。我的订单是基于购物车的，所以制作一个新的订单我将送出一个请求给端点并且我会带上request.url,这个Url是一个cart_id。所以我在这里移除了购物车。

在67-70行，我也调用了产品服务。产品服务的职责是管理产品的存储。所以我将会在这几行更改产品库存。

如果所有的流程都成功了并没有任何异常或问题。事务将会被提交。但是如果失败了，事务将会回滚。

购物车服务用来移除用户的购物车

这里是在购物车服务上的事务。首先主要内容是删除用户的购物车，其次是提交最后是回滚。

async DeleteCartById(userId: string, cartId: string): Promise<DeleteResult> {
    const userCart = await this.cartsRepository.find({
      where: {
        cart_id: cartId,
      },
    });
    this.cacheManager.set(`${userId}-cart`, JSON.stringify(userCart));
    return this.cartsRepository.delete(cartId);
}

在我删除用户的购物车之前我会查询购物车并把结果放到Redis中去

 async CommitDeleteCart(userId: string): Promise<boolean> {
    try {
      await this.cacheManager.del(`${userId}-cart`);
      return true;
    } catch (e) {
      return e;
    }
}

所以，当我想要提交事务我会删除Redis中的缓存。

async RollbackDeleteCart(userId: string): Promise<boolean> {
    try {
      const userCartCache: string = await this.cacheManager.get(
        `${userId}-cart`,
      );
      const userCart = JSON.parse(userCartCache);
      await this.cartsRepository.save(userCart);
      await this.cacheManager.del(`${userId}-cart`);
      return true;
    } catch (e) {
      return e;
    }
}

但是，如果我想要回滚，我会从redis中取得数据并把数据重新存到数据库中（我用的是MySQL）。

产品服务用来用来产品的库存

下面是处理产品服务的事务，首先主要是修改产品的数量。其次是提交最后是回滚。

async StartPatchProductVariantUserStock(
    userId: string,
    productVariantId: number,
    quantity: number,
  ): Promise<ProductVariants> {
    try {
      const variant = await this.productVariantsRepository.findOne(
        productVariantId,
        {
          where: {
            quantity: MoreThan(0),
          },
        },
      );
      if (!variant) {
        throw new Error('variant is not found or empty');
      }

      await this.cacheManager.set(
        `${userId}-order-product`,
        JSON.stringify(variant),
      );
      variant.quantity -= quantity;
      if (variant.quantity <= 0) {
        throw new Error('variant is empty');
      }
      await this.productVariantsRepository.save(variant);

      return variant;
    } catch (e) {
      console.log('failed to start transaction');
      console.log(e);
      throw e;
    }
  }

首先，通过ID查找产品。如果我不能从DB查到结果就会抛出异常。接着我知识把数据存到Redis，同时更新数据库。

async CommitPatchProductVariantUserStock(userId: string): Promise<boolean> {
    try {
      await this.cacheManager.del(`${userId}-order-product`);
      return true;
    } catch (e) {
      throw e;
    }
}

所以下面是提交的进程。就是简单的移除缓存。

 async RollbackPatchProductVariantUserStock(userId: string): Promise<boolean> {
    try {
      const productVariantUserOrderCache: string = await this.cacheManager.get(
        `${userId}-order-product`,
      );

      const productVariant: ProductVariants = JSON.parse(
        productVariantUserOrderCache,
      );
      await this.productVariantsRepository.save(productVariant);

      await this.cacheManager.del(`${userId}-order-product`);
      return true;
    } catch (e) {
      console.log(e);
      throw e;
    }
  }
}

这就是产品流程中的回滚流程。

我尽可能的学习这些概念。实际上，我没有在实际的应用中尝试这些概念。我希望，这篇文章能够帮助您们加深对Sage模式的理解。如果对您们造成什么误导我会非常抱歉。谢谢您们的阅读😁。

Link to Github

Order Service: github.com/nurcahyaari…

Cart Service: github.com/nurcahyaari…

Product Service: github.com/nurcahyaari…