2. 程式人生 > 其它 >RabbitMQ .NET Core 分散式事務

RabbitMQ .NET Core 分散式事務

阿新 • • 發佈：2022-04-05

使用 .NET 5 + RabbitMQ 實現一個分散式事務，並保證最終一致性

流程為：

　　減庫存 -> 減餘額 -> 建立訂單

RabbitMQ 中建立六個佇列：

　　減庫存佇列、減庫存死信佇列

　　減餘額佇列、減餘額死信佇列

　　建立訂單佇列、建立訂單死信佇列

一個 WebAPI 用來發起流程

四個控制檯，三個用來消費各自佇列中的訊息，一個對各種的錯誤進行協調

 

訊息的可靠性

　　生產者確認

// 開啟生產者確認
Channel.ConfirmSelect();
// 傳送訊息
Channel.BasicPublish(exchange, routingKey, props, body);

 
// 生產者確認
bool isSendMsgOk = Channel.WaitForConfirms();
// 進行訊息重發
for (int i = 0; i < repeat && !isSendMsgOk; i++)
{
   Channel.BasicPublish(exchange, routingKey, props, body);
    isSendMsgOk = Channel.WaitForConfirms();
}

　　消費者確認

client.Channel.BasicAck(ea.DeliveryTag, false);

　　傳送訊息的同時把訊息持久化到資料庫中，並記錄當前狀態，到下一個環節的時候修改該訊息的狀態。

消費者異常

　　這裡就涉及到，異常後訊息重新投遞的問題了。

　　如果 NACK 那麼這個訊息會回到佇列的最上面，然後消費者在進行消費，這時候就遇見一個問題，不知道這個訊息 Retry 了幾次，因此需要一箇中間介子記錄一下這個訊息 Retry 的次數。如果超過了一個閾值就XXX處理。

　　我這裡使用的是死信佇列進行處理的，消費者異常後直接 Nack ，並把 Request 設定成 false，該訊息就會進入到對應的死信佇列中。然後又一個排程者，訂閱死信佇列，把訊息重新投遞到佇列中，並控制重試次數，如果超過閾值就XXX處理。

 

訊息的順序

　　因為整個流程都是同步進行的所以不存在順序問題

 

重複消費

　　生產者保證一定能把訊息傳送出去就行了

　　消費者需要保證業務程式碼必須冪等。執行 SQL 的時候使用 if else 判斷一下資料是否已經存在，如果不存在就執行相關的 SQL。(減庫存、減餘額的時候 Where 庫存 >= 扣減庫存)

 

 

 

生產者

RabbitMQClient _mQClient;
public OrderController(RabbitMQClient mQClient)
{
    _mQClient = mQClient;
}

[HttpPost]
public OrderDto CreateOrder(OrderDto dto)
{
    _mQClient.Publish(dto, "DeductStock_Exchange", "", true);
    return dto;
}

Startup

services.AddSingleton(typeof(RabbitMQClient));

RabbitMQClient

   public class RabbitMQClient
    {
        private readonly IConfiguration _configuration;
        private bool IsEvent = true;
        private IModel _channel;
        public RabbitMQClient(IConfiguration configuration)
        {
            _configuration = configuration;
            ConnectionFactory factory = new ConnectionFactory
            {
                UserName = _configuration["RabbitmqConfig:Username"],
                Password = _configuration["RabbitmqConfig:Password"],
                HostName = _configuration["RabbitmqConfig:Host"],
                VirtualHost = _configuration["RabbitmqConfig:VirtualHost"],
                AutomaticRecoveryEnabled = true, //網路故障自動連線恢復
            };
            Connection = factory.CreateConnection();
        }
        public IConnection Connection { get; }
        public IModel Channel
        {
            get
            {
                if (_channel == null || !_channel.IsOpen)
                {
                    _channel = Connection.CreateModel();
                }
                return _channel;
            }
        }
        public void Publish<T>(T message, string exchange, string routingKey, bool isConfirm = true, int repeat = 5)
        {
            if (isConfirm)
            {
                Channel.ConfirmSelect();
            }
            var sendBytes = Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(message));
            PublishMessage(sendBytes, exchange, routingKey, isConfirm);
            if (isConfirm)
            {
                bool isSendMsgOk = Channel.WaitForConfirms();
                for (int i = 0; i < repeat && !isSendMsgOk; i++)
                {
                    // 進行訊息重發
                    PublishMessage(sendBytes, exchange, routingKey, isConfirm);
                    isSendMsgOk = Channel.WaitForConfirms();
                }
            }
        }
        private void PublishMessage(ReadOnlyMemory<byte> body, string exchange, string routingKey, bool isConfirm = true)
        {
            IBasicProperties props = Channel.CreateBasicProperties();
            props.MessageId = Guid.NewGuid().ToString();
            Channel.BasicPublish(exchange, routingKey, props, body);
        }
    }
}

消費者+生產者

訂閱減庫存的佇列，消費成功後向減餘額的佇列中傳送訊息

    static void Main(string[] args)
    {
        Console.Title = "減庫存";
        var build = new HostBuilder();
        build.ConfigureServices((hostContext, services) =>
        {
            var configuration = new ConfigurationBuilder()
                .SetBasePath(Directory.GetCurrentDirectory())
                .AddJsonFile("appsettings.json")
                .Build();
            RabbitMQClient client = new RabbitMQClient(configuration);

            string exchangeName = "DeductStock_Exchange";
            string queueName = "DeductStock_Queue";
            string routingKey = "DeductStock_Routing";

            string dead_ExchangeName = "DeductStock_Exchange_dead";
            string dead_QueueName = "DeductStock_Queue_dead";
            string dead_RoutingKey = "DeductStock_Routing_dead";

            client.Channel.ExchangeDeclare(dead_ExchangeName, type: "fanout", durable: true, autoDelete: false);
            client.Channel.QueueDeclare(dead_QueueName, durable: true, exclusive: false, autoDelete: false);
            client.Channel.QueueBind(dead_QueueName, dead_ExchangeName, dead_RoutingKey);

            client.Channel.ExchangeDeclare(exchangeName, type: "fanout", durable: true, autoDelete: false);
            client.Channel.QueueDeclare(queueName, durable: true, exclusive: false, autoDelete: false, arguments: new Dictionary<string, object> {
                { "x-dead-letter-exchange", dead_ExchangeName },
                { "x-dead-letter-routing-key", dead_RoutingKey },
                { "x-message-ttl", 10000 }
            });
            client.Channel.QueueBind(queueName, exchangeName, routingKey);

            //事件基本消費者
            EventingBasicConsumer consumer = new EventingBasicConsumer(client.Channel);

            //接收到訊息事件
            consumer.Received += (ch, ea) =>
            {
                try
                {
                    var message = Encoding.UTF8.GetString(ea.Body.ToArray());
                    Console.WriteLine($"收到訊息： { message }");
                    OrderDto dto = JsonConvert.DeserializeObject<OrderDto>(message);

                    if (dto.Id % 10 == 1 && !dto.IsBug)
                    {
                        throw new Exception();
                    }

                    //確認該訊息已被消費，確認完成後 RabbitMQ 會刪除該訊息
                    client.Channel.BasicAck(ea.DeliveryTag, false);

                    client.Publish(dto, "DeductBalance_Exchange", "", true);
                }
                catch (Exception ex)
                {
                    Console.WriteLine("庫存異常");
                    client.Channel.BasicNack(ea.DeliveryTag, false, false);
                }
            };
            client.Channel.BasicConsume(queueName, false, consumer);
            Console.WriteLine("庫存消費者");
        }).Build().Run();
    }
}

減餘額的和建立訂單的基本一樣

最後一個排程

訂閱死信佇列，然後把死信佇列中的訊息重新投遞到佇列中，讓佇列繼續處理相關的業務

如果達到閾值，或者出現什麼問題把訊息持久化到硬碟上面

{
    var model = new DispatchModel("", "DeductStock_Exchange", "DeductStock_Queue_dead", Directory.GetCurrentDirectory() + "\\DeductStock.txt");
    var channel = Connection.CreateModel();
    EventingBasicConsumer consumer = new EventingBasicConsumer(channel);
    consumer.Received += (ch, ea) =>
    {
        var message = Encoding.UTF8.GetString(ea.Body.ToArray());
        Console.WriteLine($"收到訊息： { message }");
        OrderDto dto = JsonConvert.DeserializeObject<OrderDto>(message);
        try
        {
            channel.BasicAck(ea.DeliveryTag, false);
            if (dto.RetryCount >= 5)
            {
                File.WriteAllText(model.messageFilePath, message);
                return;
            }
            dto.RetryCount += 1;

            // 開啟發送確認
            channel.ConfirmSelect();
            var sendBytes = Encoding.UTF8.GetBytes(JsonConvert.SerializeObject(dto));
            channel.BasicPublish(model.exchangeName, model.routingKey, null, sendBytes);
            bool isSendMsgOk = channel.WaitForConfirms();

            for (int i = 0; i < 5 && !isSendMsgOk; i++)
            {
                // 進行訊息重發
                channel.BasicPublish(model.exchangeName, model.routingKey, null, sendBytes);
                isSendMsgOk = channel.WaitForConfirms();
            }
            if (!isSendMsgOk)
            {
                /// 傳送六次都沒有成功
                /// 訊息快取到本地
                File.WriteAllText(model.messageFilePath, message);
            }
        }
        catch (Exception ex)
        {
            // 確認消費
            channel.BasicAck(ea.DeliveryTag, false);
            // 訊息快取到本地
            File.WriteAllText(model.messageFilePath, message);
        }
    };
    //啟動消費者 設定為手動應答訊息
    channel.BasicConsume(model.dead_queueName, false, consumer);
}

 

相關推薦

RabbitMQ .NET Core 分散式事務

使用 .NET 5 + RabbitMQ 實現一個分散式事務，並保證最終一致性 流程為： 　　減庫存 -> 減餘額 -> 建立訂單

MassTransit .NET Core 分散式事務

使用 .NET 5 + MassTransit 8.0.1 實現一個分散式事務，並保證最終一致性   WebAPI readonly ISendEndpointProvider _sendEndpointProvider;

RabbitMQ .net core 客戶端 EasyNetQ 的使用

依賴注入 var connectionConfiguration = new ConnectionConfiguration { Hosts = new List<HostConfiguration>

.NET Core 事件匯流排,分散式事務解決方案：CAP（轉載）

背景 相信前面幾篇關於微服務的文章也介紹了那麼多了，在構建微服務的過程中確實需要這麼一個東西，即便不是在構建微服務，那麼在構建分散式應用的過程中也會遇到分散式事務的問題，那麼 CAP 就是在這樣的背

.NET Core 事件匯流排,分散式事務解決方案：CAP

本文地址：http://www.cnblogs.com/savorboard/p/cap.html 作者部落格：Savorboard 背景 相信前面幾篇關於微服務的文章也介紹了那麼多了，在構建微服務的過程中確實需要這麼一個東西，即便不是在構建微服務，那麼在

.Net Core with 微服務 - 分散式事務 - 2PC、3PC

最近比較忙，好久沒更新了。這次我們來聊一聊分散式事務。 在微服務體系下，我們的應用被分割成多個服務，每個服務都配置一個數據庫。如果我們的服務劃分的不夠完美，那麼為了完成業務會出現非常多的跨庫事務。即使

.Net Core with 微服務 - 使用 AgileDT 快速實現基於可靠訊息的分散式事務

前面對於分散式事務也講了好幾篇了（可靠訊息最終一致性 分散式事務 - TCC 分散式事務 - 2PC、3PC），但是還沒有實戰過。那麼本篇我們就來演示下如何在 .NET 環境下實現一個基於可靠訊息的分散式事務。基於可靠訊息

分散式中介軟體訊息佇列（3）---.Net Core 使用rabbitmq

一、通過install-package rabbitmq.client命令或nuget安裝rabbitmq.client包 二、rabbitmq操作

.Net Core 整合ExceptionLess分散式日誌框架之本地化部署

前言 公司目前使用的專案中關於日誌記錄這塊，之前一直都是使用的Log4net 存放於後臺檔案中的，對於異常錯誤啊，或者需要檢視一些詳情錯誤的時候感覺很不方便，要到伺服器上去開啟日誌檔案檢索錯誤，降低了解決問題的

.NET Core 使用RabbitMQ(轉)

原文地址：https://www.cnblogs.com/stulzq/p/7551819.html RabbitMQ簡介 AMQP，即Advanced Message Queuing Protocol，高階訊息佇列協議，是應用層協議的一個開放標準，為面向訊息的中介軟體設計。訊息中介軟體

Asp.net Core 3.1基於AspectCore實現AOP，實現事務、快取攔截器

最近想給我的框架加一種功能，就是比如給一個方法加一個事務的特性Attribute，那這個方法就會啟用事務處理。給一個方法加一個快取特性，那這個方法就會進行快取。

RabbitMQ使用交換機處理非同步訊息佇列------分散式事務處理案例

RabbitMQ使用交換機處理非同步訊息佇列案例的安裝環境可以參考RabbitMQ環境準備/環境搭建，

使用RabbitMQ實現分散式事務

　　基於SOA理念的微服務越來越流行，甚至一些區域網部署的專案也採用微服務架構。微服務的好處很多，但同時也帶來了很多新的問題，分散式事務便是其中一個，出現問題後自然也會出現解決辦法，比如兩段提交、三段提交

.Net Core 快取方式（二）分散式快取 MemoryDistributedCache 實現（1）

.Net Core 快取方式（二）分散式快取 官方文件 官方文件 分散式快取是由多個應用伺服器共享的快取，通常作為外部服務在訪問它的應用伺服器上維護。 分散式快取可以提高 ASP.NET Core 應用程式的效能和可伸縮性，尤

.Net Core 快取方式（二）分散式快取的擴充套件方法的實現

.Net Core 快取方式（二）分散式快取的擴充套件方法的實現 IDistributedCache 介面 看過 IDistributedCache.cs 瞭解到，所有分散式快取都是通過實現 IDistributedCache，實現使用不同的快取伺服器，

.Net Core 快取方式（二）自定義分散式快取的擴充套件方法（5）

.Net Core 快取方式（二）自定義分散式快取的擴充套件方法（5） IDistributedCache 的擴充套件方法

.NET Core 3.x 基於AspectCore實現AOP，實現事務、快取攔截器

最近想給我的框架加一種功能，就是比如給一個方法加一個事務的特性Attribute，那這個方法就會啟用事務處理。給一個方法加一個快取特性，那這個方法就會進行快取。這個也是網上說的面向切面程式設計AOP。

.NET Core 下使用 RabbitMQ

介紹 RabbitMQ是一個開源的,基於AMQP(Advanced Message Queuing Protocol)協議的完整,可複用的企業級訊息佇列(Message Queue 一種應用程式與應用程式之間的一種通訊方法)系統,RabbitMQ可以實現點對點,釋出訂閱等訊息

.net core使用rabbitmq訊息佇列 (二)

　　之前有寫過.net core整合使用rabbitmq的博文，見.net core使用rabbitmq訊息佇列，但是裡面的使用很簡單，而且還有幾個bug，想改下，但是後來想了想，還是算了，之前使用的是.net core 2.x，還是重新整理一遍吧！

Asp.net Core CAP Kafka 分散式訊息佇列

CAP 是一個基於 .NET Standard 的 C# 庫，它是一種處理分散式事務的解決方案，同樣具有 EventBus 的功能，它具有輕量級、易使用、高效能等特點。