spring cloud 实现的event sourcing源码解读

源码项目地址

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

在cart购物车下单之后，

关于order订单模块是如何用事件溯源来重塑当前订单状态

订单的实体类

可以看到订单状态orderStatus并未持久化到db中；

@Document
public class Order extends BaseEntity {

@Id
private ObjectId orderId;
private String accountNumber;
@Transient
private OrderStatus orderStatus;
private List<LineItem> lineItems = new ArrayList<>();
private Address shippingAddress;

...

而是依靠

订单的事件实体类

这个订单事件的实体类来重塑订单的当前状态，订单事件表保存着订单从创建开始的所有发生的事件，在这个微服务模块中它是持久化到mongodb中；

@Document
public class OrderEvent extends BaseEntity implements Serializable {

private String id;
private OrderEventType type;
private String orderId;

...

这是获取当前订单的代码

public Order getOrder(String orderId, Boolean validate) {
// Get the order for the event
Order order = orderRepository.findOne(orderId);

if (validate) {
try {
// Validate the account number of the event's order belongs to the user
validateAccountNumber(order.getAccountNumber());
} catch (Exception ex) {
return null;
}
}

Flux<OrderEvent> orderEvents =
Flux.fromStream(orderEventRepository.findOrderEventsByOrderId(order.getOrderId()));

// Aggregate the state of order
return orderEvents
.takeWhile(orderEvent -> orderEvent.getType() != OrderEventType.DELIVERED)
.reduceWith(() -> order, Order::incorporate)
.get();
}

下面来一一解读，

Flux.fromStream(orderEventRepository.findOrderEventsByOrderId(order.getOrderId())):

从订单事件仓库获取当前订单的事件集合，以stream流的形式返回，包装成Flux

takeWhile(orderEvent -> orderEvent.getType() != OrderEventType.DELIVERED):

将订单事件不是‘已送达’的事件，按发生时间取出

reduceWith(() -> order, Order::incorporate):

用构造函数初始化一个订单实例，把事件按顺序交由Order::incorporate的方法去重放订单的事件来获得当前状态的订单，订单的incorporate方法实质上就是是个一个业务状态的状态机。

public Order incorporate(OrderEvent orderEvent) {

if (orderStatus == null)
orderStatus = OrderStatus.PURCHASED;

switch (orderStatus) {
case PURCHASED:
if (orderEvent.getType() == OrderEventType.CREATED)
orderStatus = OrderStatus.PENDING;
break;
case PENDING:
if (orderEvent.getType() == OrderEventType.ORDERED) {
orderStatus = OrderStatus.CONFIRMED;
} else if (orderEvent.getType() == OrderEventType.PURCHASED) {
orderStatus = OrderStatus.PURCHASED;
}
break;
case CONFIRMED:
if (orderEvent.getType() == OrderEventType.SHIPPED) {
orderStatus = OrderStatus.SHIPPED;
} else if (orderEvent.getType() == OrderEventType.CREATED) {
orderStatus = OrderStatus.PENDING;
}
break;
case SHIPPED:
if (orderEvent.getType() == OrderEventType.DELIVERED) {
orderStatus = OrderStatus.DELIVERED;
} else if (orderEvent.getType() == OrderEventType.ORDERED) {
orderStatus = OrderStatus.CONFIRMED;
}
break;
case DELIVERED:
if (orderEvent.getType() == OrderEventType.SHIPPED)
orderStatus = OrderStatus.SHIPPED;
break;
default:
// Invalid event type with regards to the order status
break;
}

return this;
}