前言
Saga單詞翻譯過來是指尤指古代挪威或冰島講述冒險經歷和英雄業績的長篇故事，對，這里強調長篇故事，許多系統都存在長時間運行的業務流程，NServiceBus使用基于事件驅動的體系結構將容錯性和可伸縮性融入這些業務處理程序中，
當然一個單一介面呼叫則算不上一個長時間運行的業務場景，那么如果在給定的用例中有兩個或多個呼叫，則應該考慮資料一致性的問題，這里有可能第一個介面呼叫成功，第二次呼叫則可能失敗或者超時，Saga的設計以簡單而健壯的方式處理這樣的業務用例，
認識Saga
先來通過一段代碼簡單認識一下Saga，在NServiceBus里，使用Saga的話則需要實作抽象類Saga
public class Saga:Saga<State>,
IAmStartedByMessages<StartOrder>,
IHandleMessages<CompleteOrder>
{
protected override void ConfigureHowToFindSaga(SagaPropertyMapper<State> mapper)
{
mapper.ConfigureMapping<StartOrder>(message=>message.OrderId).ToSaga(saga=>saga.OrderId);
mapper.ConfigureMapping<CompleteOrder>(message=>message.OrderId).ToSaga(saga=>saga.OrderId);
}
public Task Handle(StartOrder message, IMessageHandlerContext context)
{
return Task.CompletedTask;
}
public Task Handle(CompleteOrder message, IMessageHandlerContext context)
{
MarkAsComplete();
return Task.CompletedTask;
}
}
臨時狀態
長時間運行則意味著有狀態，任何涉及多個網路呼叫的行程都需要一個臨時狀態，這個臨時狀態可以存盤在記憶體中，序列化在磁盤中，也可以存盤在分布式快取中，在NServiceBus中我們定義物體，繼承抽象類ContainSagaData即可，默認情況下，所有公開訪問的屬性都會被持久化，
public class State:ContainSagaData
{
public Guid OrderId { get; set; }
}
添加行為
在NServiceBus里，處理訊息的有兩種介面：IHandlerMessages
開啟一個Saga
在前面的代碼片段里，我們看到已經實作了介面IAmStartedByMessages
處理無序訊息
如果你的業務用例中確實存在無序訊息的情況，則還需要業務流程正常輪轉，那么則需要多個messaeg都要事先介面IAmStartedByMessages介面，也就是說多個message都可以創建Saga實體，
依賴可恢復性
在處理無序訊息和多個訊息型別的時候，就存在訊息丟失的可能，必須在你的Saga狀態完成以后，這個Saga實體又收到一條訊息，但這時Saga狀態已經是完結狀態，這條訊息則仍然需要處理，這里則實作NServiceBus的IHandleSagaNotFound介面，
public class SagaNotFoundHandler:IHandleSagaNotFound
{
public Task Handle(object message, IMessageProcessingContext context)
{
return context.Reply(new SagaNotFoundMessage());
}
}
public class SagaNotFoundMessage
{
}
結束Saga
當你的業務用例不再需要Saga實體時，則呼叫MarkComplete()來結束Saga實體，這個方法在前面的代碼片段中也可以看到，其實本質也就是設定Saga.Complete屬性，這是個bool值，你在業務用例中也可以用此值來判斷Saga流程是否結束，
namespace NServiceBus
{
using System;
using System.Threading.Tasks;
using Extensibility;
public abstract class Saga
{
/// <summary>
/// The saga's typed data.
/// </summary>
public IContainSagaData Entity { get; set; }
public bool Completed { get; private set; }
internal protected abstract void ConfigureHowToFindSaga(IConfigureHowToFindSagaWithMessage sagaMessageFindingConfiguration);
protected Task RequestTimeout<TTimeoutMessageType>(IMessageHandlerContext context, DateTime at) where TTimeoutMessageType : new()
{
return RequestTimeout(context, at, new TTimeoutMessageType());
}
protected Task RequestTimeout<TTimeoutMessageType>(IMessageHandlerContext context, DateTime at, TTimeoutMessageType timeoutMessage)
{
if (at.Kind == DateTimeKind.Unspecified)
{
throw new InvalidOperationException("Kind property of DateTime 'at' must be specified.");
}
VerifySagaCanHandleTimeout(timeoutMessage);
var options = new SendOptions();
options.DoNotDeliverBefore(at);
options.RouteToThisEndpoint();
SetTimeoutHeaders(options);
return context.Send(timeoutMessage, options);
}
protected Task RequestTimeout<TTimeoutMessageType>(IMessageHandlerContext context, TimeSpan within) where TTimeoutMessageType : new()
{
return RequestTimeout(context, within, new TTimeoutMessageType());
}
protected Task RequestTimeout<TTimeoutMessageType>(IMessageHandlerContext context, TimeSpan within, TTimeoutMessageType timeoutMessage)
{
VerifySagaCanHandleTimeout(timeoutMessage);
var sendOptions = new SendOptions();
sendOptions.DelayDeliveryWith(within);
sendOptions.RouteToThisEndpoint();
SetTimeoutHeaders(sendOptions);
return context.Send(timeoutMessage, sendOptions);
}
protected Task ReplyToOriginator(IMessageHandlerContext context, object message)
{
if (string.IsNullOrEmpty(Entity.Originator))
{
throw new Exception("Entity.Originator cannot be null. Perhaps the sender is a SendOnly endpoint.");
}
var options = new ReplyOptions();
options.SetDestination(Entity.Originator);
context.Extensions.Set(new AttachCorrelationIdBehavior.State { CustomCorrelationId = Entity.OriginalMessageId });
options.Context.Set(new PopulateAutoCorrelationHeadersForRepliesBehavior.State
{
SagaTypeToUse = null,
SagaIdToUse = null
});
return context.Reply(message, options);
}
//這個方法結束saga流程，標記Completed屬性
protected void MarkAsComplete()
{
Completed = true;
}
void VerifySagaCanHandleTimeout<TTimeoutMessageType>(TTimeoutMessageType timeoutMessage)
{
var canHandleTimeoutMessage = this is IHandleTimeouts<TTimeoutMessageType>;
if (!canHandleTimeoutMessage)
{
var message = $"The type '{GetType().Name}' cannot request timeouts for '{timeoutMessage}' because it does not implement 'IHandleTimeouts<{typeof(TTimeoutMessageType).FullName}>'";
throw new Exception(message);
}
}
void SetTimeoutHeaders(ExtendableOptions options)
{
options.SetHeader(Headers.SagaId, Entity.Id.ToString());
options.SetHeader(Headers.IsSagaTimeoutMessage, bool.TrueString);
options.SetHeader(Headers.SagaType, GetType().AssemblyQualifiedName);
}
}
}
Saga持久化
本機開發環境我們使用LearningPersistence，但是投產的話則需要使用資料庫持久化，這里我們基于MySQL，SQL持久化需要引入NServiceBus.Persistence.Sql，SQL Persistence會生成幾種關系型資料庫的sql scripts，然后會根據你的斷言配置選擇所需資料庫，比如SQL Server、MySQL、PostgreSQL、Oracle，
持久化Saga自動創建所需表結構，你只需手動配置即可，配置后編譯成功后專案執行目錄下會生成sql腳本，檔案夾名稱是NServiceBus.Persistence.Sql，下面會有Saga子目錄，
/* TableNameVariable */
set @tableNameQuoted = concat('`', @tablePrefix, 'Saga`');
set @tableNameNonQuoted = concat(@tablePrefix, 'Saga');
/* Initialize */
drop procedure if exists sqlpersistence_raiseerror;
create procedure sqlpersistence_raiseerror(message varchar(256))
begin
signal sqlstate
'ERROR'
set
message_text = message,
mysql_errno = '45000';
end;
/* CreateTable */
set @createTable = concat('
create table if not exists ', @tableNameQuoted, '(
Id varchar(38) not null,
Metadata json not null,
Data json not null,
PersistenceVersion varchar(23) not null,
SagaTypeVersion varchar(23) not null,
Concurrency int not null,
primary key (Id)
) default charset=ascii;
');
prepare script from @createTable;
execute script;
deallocate prepare script;
/* AddProperty OrderId */
select count(*)
into @exist
from information_schema.columns
where table_schema = database() and
column_name = 'Correlation_OrderId' and
table_name = @tableNameNonQuoted;
set @query = IF(
@exist <= 0,
concat('alter table ', @tableNameQuoted, ' add column Correlation_OrderId varchar(38) character set ascii'), 'select \'Column Exists\' status');
prepare script from @query;
execute script;
deallocate prepare script;
/* VerifyColumnType Guid */
set @column_type_OrderId = (
select concat(column_type,' character set ', character_set_name)
from information_schema.columns
where
table_schema = database() and
table_name = @tableNameNonQuoted and
column_name = 'Correlation_OrderId'
);
set @query = IF(
@column_type_OrderId <> 'varchar(38) character set ascii',
'call sqlpersistence_raiseerror(concat(\'Incorrect data type for Correlation_OrderId. Expected varchar(38) character set ascii got \', @column_type_OrderId, \'.\'));',
'select \'Column Type OK\' status');
prepare script from @query;
execute script;
deallocate prepare script;
/* WriteCreateIndex OrderId */
select count(*)
into @exist
from information_schema.statistics
where
table_schema = database() and
index_name = 'Index_Correlation_OrderId' and
table_name = @tableNameNonQuoted;
set @query = IF(
@exist <= 0,
concat('create unique index Index_Correlation_OrderId on ', @tableNameQuoted, '(Correlation_OrderId)'), 'select \'Index Exists\' status');
prepare script from @query;
execute script;
deallocate prepare script;
/* PurgeObsoleteIndex */
select concat('drop index ', index_name, ' on ', @tableNameQuoted, ';')
from information_schema.statistics
where
table_schema = database() and
table_name = @tableNameNonQuoted and
index_name like 'Index_Correlation_%' and
index_name <> 'Index_Correlation_OrderId' and
table_schema = database()
into @dropIndexQuery;
select if (
@dropIndexQuery is not null,
@dropIndexQuery,
'select ''no index to delete'';')
into @dropIndexQuery;
prepare script from @dropIndexQuery;
execute script;
deallocate prepare script;
/* PurgeObsoleteProperties */
select concat('alter table ', table_name, ' drop column ', column_name, ';')
from information_schema.columns
where
table_schema = database() and
table_name = @tableNameNonQuoted and
column_name like 'Correlation_%' and
column_name <> 'Correlation_OrderId'
into @dropPropertiesQuery;
select if (
@dropPropertiesQuery is not null,
@dropPropertiesQuery,
'select ''no property to delete'';')
into @dropPropertiesQuery;
prepare script from @dropPropertiesQuery;
execute script;
deallocate prepare script;
/* CompleteSagaScript */
生成的表結構：
持久化配置
Saga持久化需要依賴NServiceBus.Persistence.Sql，引入后需要實作SqlSaga抽象類，抽象類需要重寫ConfigureMapping，配置Saga作業流程業務主鍵，
public class Saga:SqlSaga<State>,
IAmStartedByMessages<StartOrder>
{
protected override void ConfigureMapping(IMessagePropertyMapper mapper)
{
mapper.ConfigureMapping<StartOrder>(message=>message.OrderId);
}
protected override string CorrelationPropertyName => nameof(StartOrder.OrderId);
public Task Handle(StartOrder message, IMessageHandlerContext context)
{
Console.WriteLine($"Receive message with OrderId:{message.OrderId}");
MarkAsComplete();
return Task.CompletedTask;
}
}
static async Task MainAsync()
{
Console.Title = "Client-UI";
var configuration = new EndpointConfiguration("Client-UI");
//這個方法開啟自動建表、自動創建RabbitMQ佇列
configuration.EnableInstallers();
configuration.UseSerialization<NewtonsoftSerializer>();
configuration.UseTransport<LearningTransport>();
string connectionString = "server=127.0.0.1;uid=root;pwd=000000;database=nservicebus;port=3306;AllowUserVariables=True;AutoEnlist=false";
var persistence = configuration.UsePersistence<SqlPersistence>();
persistence.SqlDialect<SqlDialect.MySql>();
//配置mysql連接串
persistence.ConnectionBuilder(()=>new MySqlConnection(connectionString));
var instance = await Endpoint.Start(configuration).ConfigureAwait(false);
var command = new StartOrder()
{
OrderId = Guid.NewGuid()
};
await instance.SendLocal(command).ConfigureAwait(false);
Console.ReadKey();
await instance.Stop().ConfigureAwait(false);
}
Saga Timeouts
在訊息驅動型別的環境中，雖然傳遞的無連接特性可以防止在線等待程序中消耗資源，但是畢竟等待時間需要有一個上線，在NServiceBus里已經提供了Timeout方法，我們只需訂閱即可，可以在你的Handle方法中根據需要訂閱Timeout，可參考如下代碼：
public class Saga:Saga<State>,
IAmStartedByMessages<StartOrder>,
IHandleMessages<CompleteOrder>,
IHandleTimeouts<TimeOutMessage>
{
public Task Handle(StartOrder message, IMessageHandlerContext context)
{
var model=new TimeOutMessage();
//訂閱超時訊息
return RequestTimeout(context,TimeSpan.FromMinutes(10));
}
public Task Handle(CompleteOrder message, IMessageHandlerContext context)
{
MarkAsComplete();
return Task.CompletedTask;
}
protected override string CorrelationPropertyName => nameof(StartOrder.OrderId);
public Task Timeout(TimeOutMessage state, IMessageHandlerContext context)
{
//處理超時訊息
}
protected override void ConfigureHowToFindSaga(SagaPropertyMapper<State> mapper)
{
mapper.ConfigureMapping<StartOrder>(message=>message.OrderId).ToSaga(saga=>saga.OrderId);
mapper.ConfigureMapping<CompleteOrder>(message=>message.OrderId).ToSaga(saga=>saga.OrderId);
}
}
//從Timeout的原始碼看，這個方法是通過設定SendOptions，然后再把當前這個訊息發送給自己來實作
protected Task RequestTimeout<TTimeoutMessageType>(IMessageHandlerContext context, TimeSpan within, TTimeoutMessageType timeoutMessage)
{
VerifySagaCanHandleTimeout(timeoutMessage);
var sendOptions = new SendOptions();
sendOptions.DelayDeliveryWith(within);
sendOptions.RouteToThisEndpoint();
SetTimeoutHeaders(sendOptions);
return context.Send(timeoutMessage, sendOptions);
}
總結
NServiceBus因為是商業產品，對分布式訊息系統所涉及到的東西都做了實作，包括分布式事務（Outbox）、DTC都有，還有心跳檢測，監控都有，全而大，目前我們用到的也只是NServiceBus里很小的一部分功能，
轉載請註明出處，本文鏈接：https://www.uj5u.com/net/89755.html
標籤：.NET Core