基于服务的分布式事务（下）

在文章开头我们给大家介绍了一个典型的分布式事务业务场景，其中涉及到一个初始服务，以及两个参与服务。为了方便大家理解，我们将以 ServiceComb PackSpring Demo  ^[4] 为例介绍如何使用Saga实现分布式事务。

这里预订服务(Booking) 相当于之前提到的分布式事务初始服务，对外提供一个租车(Car)和酒店(Hotel)聚合服务，在 BookingController  ^[5] 中使用Spring提供的 RestTemplate  ^[6] 向租车和酒店服务转发请求。租车和酒店服务分别作为Saga事务参与方参与整个事务。预订服务、租车和酒店服务都是基于 Spring-Boot  ^[7] 编写独立进程应用，应用代码通过 @EnableOmega  ^[8] 加载Omega相关的 配置  ^[9] ，同时需要在Spring的配置文件中配置与Alpha服务相关的信息。

import org.apache.servicecomb.pack.omega.spring.EnableOmega;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
@EnableOmega
public class Application {
  public static void main(String[] args) {
    SpringApplication.run(Application.class, args);
  }
}

在应用代码中需要描述出Saga事务的边界，我们可以在BookingController的order方法上标准@SagaStart；

import org.apache.servicecomb.pack.omega.context.annotations.SagaStart;

@SagaStart
@PostMapping("/booking/{name}/{rooms}/{cars}")
public String order(@PathVariable String name, @PathVariable Integer rooms, @PathVariable Integer cars) {
  // Calling the car service
  template.postForEntity(
      carServiceUrl + "/order/{name}/{cars}",
      null, String.class, name, cars);
  // Calling the hotel service
  template.postForEntity(
      hotelServiceUrl + "/order/{name}/{rooms}",
      null, String.class, name, rooms);
  return name + " booking " + rooms + " rooms and " + cars + " cars OK";
}

本地事务是通过@Compensable来标识，并且在Compensable的compensationMethod属性中描述补偿方法。注意补偿方法和本地事务方法的参数必须一致，否则Omega在进行恢复操作的时候会报找不到方法的错误。

import org.apache.servicecomb.pack.omega.transaction.annotations.Compensable;

@Compensable(compensationMethod = "cancel")
void order(CarBooking booking) {
  booking.confirm();
  bookings.put(booking.getId(), booking);
}

void cancel(CarBooking booking) {
  Integer id = booking.getId();
  if (bookings.containsKey(id)) {
    bookings.get(id).cancel();
  }
}

下面我们会结合ServiceComb中的 TCC示例  ^[10] ，向大家详解如何编写与TCC应用代码。这个示例以我们常见的电商场景为例，用户通过Ordering应用进行下单，Odering会调用Inventory以及Payment两个服务进行相关的业务操作。和之前Saga的示例一样，我们可以通过@EnableOmega的方式在这几个应用中注入Omega。

import org.apache.servicecomb.pack.omega.spring.EnableOmega;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
@EnableOmega
public class TccOrderingApplication {
  public static void main(String[] args) {
    SpringApplication.run(TccOrderingApplication.class, args);
  }
}

接下来我们需要在OrderingController中通过加入@TccStart来定义这个TCC分布式事务的范围，这个分布式事务的范围就是order方法，order方法会调用Inventory的order服务接口，以及Payment的pay服务接口。

import org.apache.servicecomb.pack.omega.context.annotations.TccStart;

@TccStart
@PostMapping("/order/{userName}/{productName}/{productUnit}/{unitPrice}")
@ResponseBody
public String order(
    @PathVariable String userName,
    @PathVariable String productName, @PathVariable Integer productUnit, @PathVariable Integer unitPrice) {
  // Calling the inventory service
  restTemplate.postForEntity(
      inventoryServiceUrl + "/order/{userName}/{productName}/{productUnit}",
      null, String.class, userName, productName, productUnit);

  int amount = productUnit * unitPrice;
  // Calling the payment service
  restTemplate.postForEntity(paymentServiceUrl + "/pay/{userName}/{amount}",
      null, String.class, userName, amount);

  return userName + " ordering " + productName + " with " + productUnit + " OK";
}

Inventory会在try阶段先进行库存的扣减，在分布式业务执行成功之后进行设置库存订单状态；如果业务执行失败，Inventory服务会执行恢复操作。通过定义@Participate，ServiceComb Omega可以标注相关try方法，同时通过confirmMethod以及cancelMethod定义相关确认和取消方法名。需要注意的是，这里提到的confirm、cancel方法的参数必须和try方法的相同。

import org.apache.servicecomb.pack.omega.transaction.annotations.Participate;
import org.springframework.transaction.annotation.Transactional;

@Participate(confirmMethod = "confirm", cancelMethod = "cancel")
@Transactional
public void pay(Payment payment) {
  Account account = getAccount(payment);
  if (account.getCredit() >= payment.getAmount()) {
    account.setCredit(account.getCredit() - payment.getAmount());
    accountDao.saveAndFlush(account);
    payments.put(payment.getId(), payment);
  } else {
    throw new IllegalArgumentException("Insufficient funds!");
  }
}

@Transactional
Account getAccount(Payment payment) {
  Account account = accountDao.findByUserName(payment.getUserName());
  if (Objects.isNull(account)) {
    throw new IllegalArgumentException("Cannot find the account!");
  }
  return account;
}

@Transactional
public void confirm(Payment payment) {
  Account account = getAccount(payment);
  payment.setConfirmed(true);
  payment.setCancelled(false);
  account.setBalance(account.getBalance() - payment.getAmount());
  payment.setBalance(account.getBalance());
  accountDao.saveAndFlush(account);
}

@Transactional
public void cancel(Payment payment) {
  Account account = getAccount(payment);
  account.setCredit(account.getCredit() + payment.getAmount());
  accountDao.saveAndFlush(account);
  payment.setBalance(account.getBalance());
  payment.setConfirmed(false);
  payment.setCancelled(true);
}

Payment会在try阶段先验证用户账户的信用值，在确认阶段执行扣减用户账户余额的操作，在恢复阶段恢复信用值。

import org.apache.servicecomb.pack.omega.transaction.annotations.Participate;
import org.springframework.transaction.annotation.Transactional;

@Participate(confirmMethod = "confirm", cancelMethod = "cancel")
@Transactional
public void reserve(ProductOrder order) {
  Product product = getProduct(order.getProductName());
  if (product.getInStock() >= order.getUnits()) {
    product.setInStock(product.getInStock() - order.getUnits());
    productDao.saveAndFlush(product);
    orders.put(order.getId(), order);
  } else {
    throw new IllegalArgumentException("The Product is out of stock!");
  }
}

public void confirm(ProductOrder order) {
  order.setConfirmed(true);
}

@Transactional
public void cancel(ProductOrder order) {
  Product product = productDao.findProduceByName(order.getProductName());
  product.setInStock(product.getInStock() + order.getUnits());
  productDao.saveAndFlush(product);
  order.setCancelled(true);
}

@Transactional
private Product getProduct(String productName) {
  Product product = productDao.findProduceByName(productName);
  if (Objects.isNull(product)) {
    throw new IllegalArgumentException("Product not exists at all!");
  }
  return product;
}

在本文中我们先从分布式事务场景入手，采用交互图的方式，向大家介绍分布式Saga以及TCC分布式事务协调协议交互，最后结合ServiceComb Pack所提供的示例向大家介绍如何编写Saga以及TCC的应用代码。

参考链接详情

[1]ServiceComb Pack

https://github.com/apache/servicecomb-pack

[2]Sagas

https://www.cs.cornell.edu/andru/cs711/2002fa/reading/sagas.pdf

[3]OmegaContext

https://willemjiang.github.io/servicecomb/pack/2019/01/10/[OmegaContext](https://github.com/apache/servicecomb-packblob/master/omega/omega-context/src/main/java/org/apache/servicecomb/saga/omega/context/OmegaContext.java)

[4]ServiceComb Pack Spring Demo

https://github.com/apache/servicecomb-pack/tree/master/demo/saga-spring-demo

[5]BookingController

https://github.com/apache/servicecomb-pack/blob/master/demo/saga-spring-demo/booking/src/main/java/org/apache/servicecomb/pack/demo/booking/BookingController.java

[6]RestTemplate

https://docs.spring.io/spring/docs/current/javadoc-api/org/springframework/web/client/RestTemplate.html

[7] Spring-Boot

https://spring.io/projects/spring-boot

[8]@EnableOmega

https://github.com/apache/servicecomb-pack/blob/master/omega/omega-spring-starter/src/main/java/org/apache/servicecomb/pack/omega/spring/EnableOmega.java

[9]配置

https://github.com/apache/servicecomb-pack/blob/master/demo/saga-spring-demo/booking/src/main/resources/application.yaml

[10]TCC示例

https://github.com/apache/servicecomb-pack/tree/master/demo/tcc-spring-demo