C#通過rabbitmq實(shí)現(xiàn)定時(shí)任務(wù)(延時(shí)隊(duì)列)

更新時(shí)間：2021年05月28日 09:39:24 作者：初夏的陽光丶

工作中經(jīng)常會(huì)有定時(shí)任務(wù)的需求,常見的做法可以使用Timer、Quartz、Hangfire等組件，本文使用C#通過rabbitmq實(shí)現(xiàn)定時(shí)任務(wù)(延時(shí)隊(duì)列)，感興趣的可以了解一下

本文主要講解如何通過RabbitMQ實(shí)現(xiàn)定時(shí)任務(wù)(延時(shí)隊(duì)列)

環(huán)境準(zhǔn)備

需要在MQ中進(jìn)行安裝插件地址鏈接
插件介紹地址：https://www.rabbitmq.com/blog/2015/04/16/scheduling-messages-with-rabbitmq/

使用場(chǎng)景

作為一個(gè)新的預(yù)支付訂單被初始化放置，如果該訂單在指定時(shí)間內(nèi)未進(jìn)行支付，則將被認(rèn)為超時(shí)訂單進(jìn)行關(guān)閉處理；電商系統(tǒng)中應(yīng)用較多，用戶購(gòu)買商品產(chǎn)生訂單，但未進(jìn)行支付，訂單產(chǎn)生30分鐘內(nèi)未支付將關(guān)閉訂單（且滿足該場(chǎng)景數(shù)量龐大），不可能采用人工干預(yù)。

代碼介紹

生產(chǎn)者

   var factory = new ConnectionFactory()
            {
                Uri = new Uri("MQ地址")
            };
           
            using var connection = factory.CreateConnection();
            using var channel = connection.CreateModel();


            var exchangeName = "delay-exchange";
            var routingkey = "delay.delay";
            var queueName = "delay_queueName";
            //設(shè)置Exchange隊(duì)列類型
            var argMaps = new Dictionary<string, object>()
            {
                {"x-delayed-type", "topic"}
            };
            //設(shè)置當(dāng)前消息為延時(shí)隊(duì)列
            channel.ExchangeDeclare(exchange: exchangeName, type: "x-delayed-message", true, false, argMaps);
            channel.QueueDeclare(queueName, true, false, false, argMaps);
            channel.QueueBind(queueName, exchangeName, routingkey);
            for (int i = 0; i < 3; i++)
            {
                var time = 1000 * 5;
                var message = $@"發(fā)送時(shí)間為 {DateTime.Now:yyyy-MM-dd HH:mm:ss} 延時(shí)時(shí)間為:{time}";
                var body = Encoding.UTF8.GetBytes(message);
                var props = channel.CreateBasicProperties();
                //設(shè)置消息的過期時(shí)間
                props.Headers = new Dictionary<string, object>()
                {
                    {  "x-delay", 5000 }
                };
                channel.BasicPublish(exchange: exchangeName,
                    routingKey: routingkey,
                    basicProperties: props,
                    body: body);
                Console.WriteLine(message);


            }
            Console.ReadLine();

消費(fèi)者(自動(dòng)綁定隊(duì)列寫法)

        var factory = new ConnectionFactory()
            {
                Uri = new Uri(MQ地址)
            };
            using var connection = factory.CreateConnection();
            using var channel = connection.CreateModel();
            var queueName = "delay_queueName";
            channel.QueueDeclare(queueName, true, false, false, null);
            var consumer = new EventingBasicConsumer(channel);
            consumer.Received += (model, ea) =>
            {
                var body = ea.Body;
                var message = Encoding.UTF8.GetString(body);
                var routingKey = ea.RoutingKey;
                Console.WriteLine($@"接受到消息的時(shí)間為 {DateTime.Now:yyyy-MM-dd HH:mm:ss},routingKey:{routingKey} message:{message} ");
            };
            channel.BasicConsume(queue: queueName,
                autoAck: true,
                consumer: consumer);
            Console.ReadLine();

消費(fèi)者(手動(dòng)綁定隊(duì)列寫法)

  var factory = new ConnectionFactory()
            {
                Uri = new Uri(MQ地址)
            };
             using var connection = factory.CreateConnection();
            using var channel = connection.CreateModel();
            var exchangeName = "delay-exchange";
            var routingkey = "delay.delay";
            var queueName = "delay_queueName";
            var autoDelete = true;

            var argMaps = new Dictionary<string, object>()
            {
                {"x-delayed-type", "topic"}
            };
            channel.ExchangeDeclare(exchange: exchangeName, type: "x-delayed-message", true, false, argMaps);
            channel.QueueDeclare(queueName, true, false, false, argMaps);
            channel.QueueBind(queue: queueName, exchange: exchangeName, routingKey: routingkey);
            //channel.QueueDeclare(queueName, true, false, false, null);
            var consumer = new EventingBasicConsumer(channel);
            consumer.Received += (model, ea) =>
            {
                var body = ea.Body;
                var message = Encoding.UTF8.GetString(body);
                var routingKey = ea.RoutingKey;
                Console.WriteLine($@"接受到消息的時(shí)間為 {DateTime.Now:yyyy-MM-dd HH:mm:ss},routingKey:{routingKey} message:{message} ");
            };
            channel.BasicConsume(queue: queueName,
                autoAck: true,
                consumer: consumer);
            Console.ReadLine();

最終實(shí)現(xiàn)效果(兩個(gè)消費(fèi)者)

在上述實(shí)現(xiàn)中，其實(shí)主要靠以下參數(shù)來幫我們實(shí)現(xiàn)當(dāng)前功能

聲明Exchange中的 type: "x-delayed-message" 這個(gè)表明當(dāng)前隊(duì)列為延時(shí)消息隊(duì)列
聲明Exchange中arguments中的 {"x-delayed-type", "topic"} 當(dāng)前表明當(dāng)前隊(duì)列為Topic模式
最后我們?cè)贑reateBasicProperties的Header中設(shè)置 { "x-delay", 5000 }來達(dá)到消息延時(shí)的功能(單位為ms)

建議

如果使用當(dāng)前模式來做定時(shí)任務(wù),在要求消息不丟失的前提下，需要運(yùn)維同學(xué)提供穩(wěn)定的MQ環(huán)境

到此這篇關(guān)于C#通過rabbitmq實(shí)現(xiàn)定時(shí)任務(wù)(延時(shí)隊(duì)列)的文章就介紹到這了,更多相關(guān)C# rabbitmq定時(shí)任務(wù)內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: