C#通过rabbitmq如何实现定时任务

发布时间:2021-05-28 10:22:50 作者:小新
来源:亿速云 阅读:200

小编给大家分享一下C#通过rabbitmq如何实现定时任务,相信大部分人都还不怎么了解,因此分享这篇文章给大家参考一下,希望大家阅读完这篇文章后大有收获,下面让我们一起去了解一下吧!

环境准备

需要在MQ中进行安装插件 地址链接
插件介绍地址:https://www.rabbitmq.com/blog/2015/04/16/scheduling-messages-with-rabbitmq/

使用场景

作为一个新的预支付订单被初始化放置,如果该订单在指定时间内未进行支付,则将被认为超时订单进行关闭处理;电商系统中应用较多,用户购买商品产生订单,但未进行支付,订单产生30分钟内未支付将关闭订单(且满足该场景数量庞大),不可能采用人工干预。

代码介绍

生产者

   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";
            //设置Exchange队列类型
            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(queueName, exchangeName, routingkey);
            for (int i = 0; i < 3; i++)
            {
                var time = 1000 * 5;
                var message = $@"发送时间为 {DateTime.Now:yyyy-MM-dd HH:mm:ss} 延时时间为:{time}";
                var body = Encoding.UTF8.GetBytes(message);
                var props = channel.CreateBasicProperties();
                //设置消息的过期时间
                props.Headers = new Dictionary<string, object>()
                {
                    {  "x-delay", 5000 }
                };
                channel.BasicPublish(exchange: exchangeName,
                    routingKey: routingkey,
                    basicProperties: props,
                    body: body);
                Console.WriteLine(message);


            }
            Console.ReadLine();

消费者(自动绑定队列写法)

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($@"接受到消息的时间为 {DateTime.Now:yyyy-MM-dd HH:mm:ss},routingKey:{routingKey} message:{message} ");
            };
            channel.BasicConsume(queue: queueName,
                autoAck: true,
                consumer: consumer);
            Console.ReadLine();

消费者(手动绑定队列写法)

  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($@"接受到消息的时间为 {DateTime.Now:yyyy-MM-dd HH:mm:ss},routingKey:{routingKey} message:{message} ");
            };
            channel.BasicConsume(queue: queueName,
                autoAck: true,
                consumer: consumer);
            Console.ReadLine();

最终实现效果(两个消费者)

C#通过rabbitmq如何实现定时任务

在上述实现中,其实主要靠以下参数来帮我们实现当前功能

声明Exchange中的 type: "x-delayed-message" 这个表明当前队列为延时消息队列
声明Exchange中arguments中的 {"x-delayed-type", "topic"} 当前表明当前队列为Topic模式
最后 我们在CreateBasicProperties的Header中设置 { "x-delay", 5000 }来达到消息延时的功能(单位为ms)

以上是“C#通过rabbitmq如何实现定时任务”这篇文章的所有内容,感谢各位的阅读!相信大家都有了一定的了解,希望分享的内容对大家有所帮助,如果还想学习更多知识,欢迎关注亿速云行业资讯频道!

推荐阅读:
  1. RabbitMQ通过shovel插件迁移数据
  2. C#中怎么操作RabbitMQ

免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。

rabbitmq

上一篇:js如何实现淘宝放大镜

下一篇:Python如何实现智慧校园一键评教

相关阅读

您好,登录后才能下订单哦!

密码登录
登录注册
其他方式登录
点击 登录注册 即表示同意《亿速云用户服务条款》