NServiceBus+Saga开发分布式应用

显示全部楼层 · 2019-10-24 09:48:39

媒介

当你在处理异步消息时，每个单独的消息处理程序都是一个单独的handler，每个handler之间互不影响。这时如果一个消息依赖另一个消息的状态呢？这时业务逻辑怎么处理？

借用我们上篇文章的业务场景，如果在Ship项目里须要发送一个ShipOrder Command。这个ShipOrder须要依赖Sales.OrderPlaced和Bill.OrderBilled Command的状态，现在我们的两个单独的Message Handler都没有保持任何的状态字段，所以这时如果我们须要完成这个业务模子，就须要跟踪他们的状态。

什么是Saga

这个就是本篇文章要提的saga，界说在NServiceBus框架里，他的本质是一个消息驱动模子里的状态机，或者也可以明白为一系列消息处理程序用来共享状态的业务模子。我明白在消息队列里如果我们要包管消息同等性通常会本身创建一张Event表，这里saga维持状态的脚色有点像我们这里的Event表。

好的，回到正题上，如果我们须要在Shipping Service里发送一个ShipOrder，发送他之前须要确定OrderPlaced和OrderBilled的状态，确保这两个消息都收到以后才华发送ShipOrder。

如何利用Saga

固然，我临时明白Saga的目的是为了处理在长时间运行的使命里包管数据同等性如许的一个脚色。

Saga状态

saga状态主要是告诉NServiceBus在处理数据同等性的判定逻辑，这里须要继承抽象类ContainSagaData，在我们这个业务场景中则主要是判定OrderPlaced和OrderBilled消息是否已经吸取到并处理。

<code>public class ShippingPolicyData:ContainSagaData
{
public string OrderId { get; set; }
public bool IsOrderPlaced { get; set; }
public bool IsOrderBilled { get; set; }
}</code>

复制代码

Saga如何工作

有了状态以后，我们还须要一个“handler”来告诉NServiceBus，在这个handler里主要用来处理消息数据同等性，我看了官方文档后，他们建议我们这里的handler脚色利用Policy后缀定名，固然我觉的也可以用Saga后缀定名，比如ShippingPolicy或者ShippingSaga。
同时这里我们这个handler觉色还要继承Saga类，Saga类主要重写方法ConfigureHowToFindSaga，这个方法的作用主要是在继承的消息和我们的Saga实体之间创建映射关系。

<code> public class ShipPolicy:Saga<ShippingPolicyData>,
IAmStartedByMessages<OrderPlaced>,
IAmStartedByMessages<OrderBilled> //都可以创建Saga实例
{
private static ILog log = LogManager.GetLogger<ShipPolicy>();
protected override void ConfigureHowToFindSaga(SagaPropertyMapper<ShippingPolicyData> mapper)
{
mapper.ConfigureMapping<OrderPlaced>(t=>t.OrderId).ToSaga(sagaData=>sagaData.OrderId);
mapper.ConfigureMapping<OrderBilled>(t=>t.OrderId).ToSaga(sagaData=>sagaData.OrderId);
}
public Task Handle(OrderPlaced message, IMessageHandlerContext context)
{
log.Info("OrderPlaced message received ");
this.Data.IsOrderPlaced = true;
return ProcessOrder(context);
}
public Task Handle(OrderBilled message, IMessageHandlerContext context)
{
log.Info("OrderBilled message received");
this.Data.IsOrderBilled = true;
return ProcessOrder(context);
}
private async Task ProcessOrder(IMessageHandlerContext context)
{
if (Data.IsOrderBilled && Data.IsOrderPlaced)
{
await context.SendLocal(new ShipOrder()
{
OrderId = Data.OrderId
});
MarkAsComplete();
}
}
}</code>

复制代码

这个类里你会发现还实现了接口IAmStartedByMessages, 这个接口主要是告诉Saga，岂论是那种消息范例先进来，都可以创建一个Saga实例，就比如是Event表，不管谁人消息进来，都须要先插入一条数据，后续消息再进来时要更新数据状态，固然，这里的Saga实例也好，Event表也好，关键问题就是有效标识，或者叫主键，我们这个业务模子里，OrderPlaced和OrderBilled都包含一个属性OrderId, 这里Saga实例则利用这个OrderId做关键属性。