.NET開源的處理分散式事務的解決方案

前言

在分散式系統中，由於各個系統服務之間的獨立性和網路通訊的不確定性，要確保跨系統的事務操作的最終一致性是一項重大的挑戰。今天給大家推薦一個.NET開源的處理分散式事務的解決方案基於 .NET Standard 的 C# 庫：CAP。

CAP專案介紹

CAP 是一個基於 .NET Standard 的 C# 庫，它是一種處理分散式事務的解決方案，同樣具有 EventBus 的功能，它具有輕量級、易使用、高效能等特點。CAP 是一個EventBus，同時也是一個在微服務或者SOA系統中解決分散式事務問題的一個框架。它有助於建立可延伸，可靠並且易於更改的微服務系統。

什麼是 EventBus？

事件匯流排是一種機制，它允許不同的元件彼此通訊而不彼此瞭解。元件可以將事件傳送到Eventbus，而無需知道是誰來接聽或有多少其他人來接聽。元件也可以偵聽Eventbus上的事件，而無需知道誰傳送了事件。這樣，元件可以相互通訊而無需相互依賴。同樣，很容易替換一個元件。只要新元件瞭解正在傳送和接收的事件，其他元件就永遠不會知道。

CAP架構預覽

CAP支援的儲存

SQL Server、MySQL、PostgreSql、MongoDB、In-Memory Storage。

CAP 支援以下幾種運輸方式

RabbitMQ、Kafka、Azure Service Bus、Amazon SQS、NATS、In-Memory Queue、Redis Streams、Apache Pulsar。

怎麼選擇運輸器

專案原始碼

快速開始

安裝DotNetCore.CAP Nuget包

CAP 支援主流的訊息佇列作為傳輸器：

我本地安裝的是DotNetCore.CAP.RabbitMQ。

//你可以按需選擇下面的包進行安裝:
PM> Install-Package DotNetCore.CAP.Kafka
PM> Install-Package DotNetCore.CAP.RabbitMQ
PM> Install-Package DotNetCore.CAP.AzureServiceBus
PM> Install-Package DotNetCore.CAP.AmazonSQS
PM> Install-Package DotNetCore.CAP.NATS
PM> Install-Package DotNetCore.CAP.RedisStreams
PM> Install-Package DotNetCore.CAP.Pulsar

CAP 提供了主流資料庫作為儲存：

我本地安裝的是DotNetCore.CAP.MongoDB。

// 按需選擇安裝你正在使用的資料庫:
PM> Install-Package DotNetCore.CAP.SqlServer
PM> Install-Package DotNetCore.CAP.MySql
PM> Install-Package DotNetCore.CAP.PostgreSql
PM> Install-Package DotNetCore.CAP.MongoDB

設定CAP到 Program.cs 檔案中，如下：

            builder.Services.AddCap(x =>
            {
                //如果你使用的 EF 進行資料操作，你需要新增如下設定：
                //設定資料庫上下文
                x.UseEntityFramework<AppDbContext>();

                //如果你使用的 MongoDB，你可以新增如下設定：
                x.UseMongoDB("ConnectionStrings");  //注意，僅支援MongoDB 4.0+叢集

                //CAP RabbitMQ 設定
                x.UseRabbitMQ(rab => {
                    rab.HostName = "192.0.1.1";
                    rab.Password = "123456";
                    rab.Port = 5672;
                    rab.UserName = "123456";
                });
            });

釋出

在 Controller 中注入 ICapPublisher 然後使用 ICapPublisher 進行訊息傳送。

public class PublishController : Controller
{
    private readonly ICapPublisher _capBus;

    public PublishController(ICapPublisher capPublisher)
    {
        _capBus = capPublisher;
    }
    
    //不使用事務
    [Route("~/without/transaction")]
    public IActionResult WithoutTransaction()
    {
        _capBus.Publish("xxx.services.show.time", DateTime.Now);

        // Publish delay message
        _capBus.PublishDelayAsync(TimeSpan.FromSeconds(delaySeconds), "xxx.services.show.time", DateTime.Now);
 
        return Ok();
    }

    //Ado.Net 中使用事務，自動提交
    [Route("~/adonet/transaction")]
    public IActionResult AdonetWithTransaction()
    {
        using (var connection = new MySqlConnection(ConnectionString))
        {
            using (var transaction = connection.BeginTransaction(_capBus, autoCommit: true))
            {
                //業務程式碼

                _capBus.Publish("xxx.services.show.time", DateTime.Now);
            }
        }
        return Ok();
    }

    //EntityFramework 中使用事務，自動提交
    [Route("~/ef/transaction")]
    public IActionResult EntityFrameworkWithTransaction([FromServices]AppDbContext dbContext)
    {
        using (var trans = dbContext.Database.BeginTransaction(_capBus, autoCommit: true))
        {
            //業務程式碼

            _capBus.Publish("xxx.services.show.time", DateTime.Now);
        }
        return Ok();
    }
}


namespace xxx.Service
{
    public interface ISubscriberService
    {
        void CheckReceivedMessage(DateTime datetime);
    }

    public class SubscriberService: ISubscriberService, ICapSubscribe
    {
        [CapSubscribe("xxx.services.show.time")]
        public void CheckReceivedMessage(DateTime datetime)
        {
        }
    }
}

專案原始碼地址

更多專案實用功能和特性歡迎前往專案開源地址檢視