【rabbitMQ】-延迟队列-模拟控制智能家居的操作指令-6ren

【rabbitMQ】-延迟队列-模拟控制智能家居的操作指令

转载作者：我是一只小鸟更新时间：2023-05-26 22:31:19
这个需求为控制智能家居工作，把控制智能家居的操作指令发到队列中，比如：扫地机、洗衣机到指定时间工作
一.什么是延迟队列？
延迟队列存储的对象是对应的延迟消息，所谓“延迟消息” 是指当消息被发送以后，并不想让消费者立刻拿到消息，而是等待特定时间后，消费者才能拿到这个消息进行消费。
。
二.如何设置延迟队列?
1.配置2个业务队列（扫地机执行命令业务队列，洗衣机执行命令业务队列），绑定到业务交换机上。
2.为业务队列配置死信交换机和路由key 。
3.为死信交换机配置 2个延迟队列（扫地机执行命令延迟队列，洗衣机执行命令延迟队列）
。
三.延迟队列代码落地。
。
                          
                            using
                          
                          
                             MengLin.Shopping.RabbitMQ.Common;

                          
                          
                            using
                          
                          
                             RabbitMQ.Client;

                          
                          
                            using
                          
                          
                             System;

                          
                          
                            using
                          
                          
                             System.Collections;

                          
                          
                            using
                          
                          
                             System.Collections.Generic;

                          
                          
                            using
                          
                          
                             System.Linq;

                          
                          
                            using
                          
                          
                             System.Text;

                          
                          
                            using
                          
                          
                             System.Threading.Tasks;


                          
                          
                            namespace
                          
                          
                             MengLin.Shopping.RabbitMQ.MsgProducer.QueueType.Delay
{
    
                          
                          
                            ///
                          
                          
                            <summary>
                          
                          
                            ///
                          
                          
                             延迟队列
    
                          
                          
                            ///
                          
                          
                            </summary>
                          
                          
                            public
                          
                          
                            class
                          
                          
                             DelayQueue
    {
        
                          
                          
                            public
                          
                          
                            static
                          
                          
                            void
                          
                          
                             SendMessage()
        {
            
                          
                          
                            //
                          
                          
                            业务交换机
                          
                          
                            var
                          
                           exchangeBusiness = 
                          
                            "
                          
                          
                            business-exchange
                          
                          
                            "
                          
                          
                            ;
            
            
                          
                          
                            var
                          
                           queueBusinessSaoDiJiCommand = 
                          
                            "
                          
                          
                            SaoDiJiCommand-business-queue
                          
                          
                            "
                          
                          ;
                          
                            //
                          
                          
                            扫地机执行命令业务队列
                          
                          
                            var
                          
                           routeKeyBusinessSaoDiJiCommand = 
                          
                            "
                          
                          
                            SaoDiJiCommand-business-routekey
                          
                          
                            "
                          
                          ;
                          
                            //
                          
                          
                            扫地机执行命令路由key
                          
                          
                            var
                          
                           queueBusinessXiYiJiCommand = 
                          
                            "
                          
                          
                            XiYiJiCommand-business-queue
                          
                          
                            "
                          
                          ;
                          
                            //
                          
                          
                            洗衣机执行命令业务队列
                          
                          
                            var
                          
                           routeKeyBusinessXiYiJiCommand = 
                          
                            "
                          
                          
                            XiYiJiCommand-business-routekey
                          
                          
                            "
                          
                          ;
                          
                            //
                          
                          
                            洗衣机执行命令路由key

            
                          
                          
                            //
                          
                          
                            死信交换机
                          
                          
                            var
                          
                           exchangeDeadLetter = 
                          
                            "
                          
                          
                            deadLetter-exchange
                          
                          
                            "
                          
                          
                            ;

            
                          
                          
                            //
                          
                          
                            扫地机执行命令的延时队列和路由key
                          
                          
                            var
                          
                           delaySaoDiJiCommandQueue = 
                          
                            "
                          
                          
                            SaoDiJiCommand-delay-queue
                          
                          
                            "
                          
                          
                            ;
            
                          
                          
                            var
                          
                           routeKeyDelaySaoDiJi = 
                          
                            "
                          
                          
                            SaoDiJi-delay-routekey
                          
                          
                            "
                          
                          
                            ;

            
                          
                          
                            //
                          
                          
                            洗衣机执行命令的延时队列和路由key
                          
                          
                            var
                          
                           delayXiYiJiCommandQueue = 
                          
                            "
                          
                          
                            XiYiJiCommand-delay-queue
                          
                          
                            "
                          
                          
                            ;
            
                          
                          
                            var
                          
                           routeKeyDelayXiYiJi = 
                          
                            "
                          
                          
                            XiYiJi-delay-routekey
                          
                          
                            "
                          
                          
                            ;

            
                          
                          
                            using
                          
                           (
                          
                            var
                          
                           connection =
                          
                             RabbitMQHelper.GetConnection())
            {
                
                          
                          
                            using
                          
                           (
                          
                            var
                          
                           channel =
                          
                             connection.CreateModel())
                {
                    
                          
                          
                            //
                          
                          
                            创建死信交换机（死信交换机对交换机类型没有要求）
                          
                          
                    channel.ExchangeDeclare(exchange: exchangeDeadLetter, type: ExchangeType.Direct, durable: 
                          
                            true
                          
                          , autoDelete: 
                          
                            false
                          
                          
                            );

                    
                          
                          
                            //
                          
                          
                            创建扫地机执行命令的延时队列
                          
                          
                    channel.QueueDeclare(queue: delaySaoDiJiCommandQueue, durable: 
                          
                            true
                          
                          , exclusive: 
                          
                            false
                          
                          , autoDelete: 
                          
                            false
                          
                          
                            );

                    
                          
                          
                            //
                          
                          
                            创建洗衣机执行命令的延时队列
                          
                          
                    channel.QueueDeclare(queue: delayXiYiJiCommandQueue, durable: 
                          
                            true
                          
                          , exclusive: 
                          
                            false
                          
                          , autoDelete: 
                          
                            false
                          
                          
                            );

                    
                          
                          
                            //
                          
                          
                            绑定死信交换机到扫地机执行命令的延时队列
                          
                          
                                                channel.QueueBind(exchange: exchangeDeadLetter, queue: delaySaoDiJiCommandQueue, routingKey: routeKeyDelaySaoDiJi);

                    
                          
                          
                            //
                          
                          
                            绑定死信交换机到洗衣机执行命令的延时队列
                          
                          
                                                channel.QueueBind(exchange: exchangeDeadLetter, queue: delayXiYiJiCommandQueue, routingKey: routeKeyDelayXiYiJi);

                    
                          
                          
                            //
                          
                          
                            创建业务交换机
                          
                          
                    channel.ExchangeDeclare(exchange: exchangeBusiness, type: ExchangeType.Direct, durable: 
                          
                            true
                          
                          , autoDelete: 
                          
                            false
                          
                          
                            );

                    
                          
                          
                            //
                          
                          
                            创建扫地机命令业务队列
                          
                          
                    Dictionary<
                          
                            string
                          
                          , 
                          
                            object
                          
                          > dic1 = 
                          
                            new
                          
                           Dictionary<
                          
                            string
                          
                          , 
                          
                            object
                          
                          >
                          
                            ();
                    dic1.Add(
                          
                          
                            "
                          
                          
                            x-expires
                          
                          
                            "
                          
                          , 
                          
                            30
                          
                           * 
                          
                            6
                          
                           * 
                          
                            10000
                          
                          ); 
                          
                            //
                          
                          
                            30分钟后扫地机命令业务队列自动干掉
                          
                          
                    dic1.Add(
                          
                            "
                          
                          
                            x-message-ttl
                          
                          
                            "
                          
                          , 
                          
                            10
                          
                           * 
                          
                            6
                          
                           * 
                          
                            10000
                          
                          );
                          
                            //
                          
                          
                            设置消息在扫地机命令业务队列中的存活时间，即过期时间（消息如果10分钟内没有消费，就会放入扫地机执行命令的延时队列里）
                          
                          
                    dic1.Add(
                          
                            "
                          
                          
                            x-dead-letter-exchange
                          
                          
                            "
                          
                          , exchangeDeadLetter);
                          
                            //
                          
                          
                            过期消息转向路由  
                          
                          
                    dic1.Add(
                          
                            "
                          
                          
                            x-dead-letter-routing-key
                          
                          
                            "
                          
                          , routeKeyDelaySaoDiJi);
                          
                            //
                          
                          
                            设置死信交换机的路由key，死信交换机会根据路由key去找到对应的延迟队列
                          
                          
                    channel.QueueDeclare(queue: queueBusinessSaoDiJiCommand, durable: 
                          
                            true
                          
                          , exclusive: 
                          
                            false
                          
                          , autoDelete: 
                          
                            false
                          
                          
                            , arguments:dic1);
                    
                          
                          
                            //
                          
                          
                            绑定业务交换机到扫地机命令业务队列
                          
                          
                                                channel.QueueBind(queue: queueBusinessSaoDiJiCommand, exchange: exchangeBusiness, routingKey: routeKeyBusinessSaoDiJiCommand);


                    
                          
                          
                            //
                          
                          
                            创建洗衣机命令业务队列
                          
                          
                    Dictionary<
                          
                            string
                          
                          , 
                          
                            object
                          
                          > dic2 = 
                          
                            new
                          
                           Dictionary<
                          
                            string
                          
                          , 
                          
                            object
                          
                          >
                          
                            ();
                    dic2.Add(
                          
                          
                            "
                          
                          
                            x-expires
                          
                          
                            "
                          
                          , 
                          
                            30
                          
                           * 
                          
                            6
                          
                           * 
                          
                            10000
                          
                          ); 
                          
                            //
                          
                          
                            30分钟后洗衣机命令业务队列自动干掉
                          
                          
                    dic2.Add(
                          
                            "
                          
                          
                            x-message-ttl
                          
                          
                            "
                          
                          , 
                          
                            20
                          
                           * 
                          
                            6
                          
                           * 
                          
                            10000
                          
                          );
                          
                            //
                          
                          
                            设置消息在洗衣机命令业务队列中的存活时间，即过期时间（消息如果20分钟内没有消费，就会放入洗衣机执行命令的延时队列里）
                          
                          
                    dic2.Add(
                          
                            "
                          
                          
                            x-dead-letter-exchange
                          
                          
                            "
                          
                          , exchangeDeadLetter);
                          
                            //
                          
                          
                            过期消息转向路由  
                          
                          
                    dic2.Add(
                          
                            "
                          
                          
                            x-dead-letter-routing-key
                          
                          
                            "
                          
                          , routeKeyDelayXiYiJi);
                          
                            //
                          
                          
                            过期消息转向路由相匹配routingkey  
                          
                          
                    channel.QueueDeclare(queue: queueBusinessXiYiJiCommand, durable: 
                          
                            true
                          
                          , exclusive: 
                          
                            false
                          
                          , autoDelete: 
                          
                            false
                          
                          
                            , arguments: dic2);
                    
                          
                          
                            //
                          
                          
                            绑定业务交换机到洗衣机命令业务队列
                          
                          
                                                channel.QueueBind(queue: queueBusinessXiYiJiCommand, exchange: exchangeBusiness, routingKey: routeKeyBusinessXiYiJiCommand);

                    Console.WriteLine(
                          
                          
                            "
                          
                          
                            请输入发送的内容:
                          
                          
                            "
                          
                          
                            );
                    
                          
                          
                            var
                          
                           message =
                          
                             Console.ReadLine();
                    
                          
                          
                            var
                          
                           body =
                          
                             Encoding.UTF8.GetBytes(message);
                  

                    
                          
                          
                            //
                          
                          
                            消息持久化，既然都要用到死信队列了，说明这条消息还是比较重要的
                          
                          
                            var
                          
                           properties =
                          
                             channel.CreateBasicProperties();
                    properties.Persistent 
                          
                          = 
                          
                            true
                          
                          
                            ;

                    
                          
                          
                            if
                          
                          (message.Contains(
                          
                            "
                          
                          
                            aa
                          
                          
                            "
                          
                          
                            ))
                    {
                        
                          
                          
                            //
                          
                          
                            发布消息到扫地机命令业务队列
                          
                          
                                                    channel.BasicPublish(exchange: exchangeBusiness,
                                             routingKey: routeKeyBusinessSaoDiJiCommand,
                                             basicProperties: properties,
                                             body: body);
                    }
                    
                          
                          
                            else
                          
                          
                            
                    {
                        
                          
                          
                            //
                          
                          
                            发布消息洗衣机命令业务队列
                          
                          
                                                    channel.BasicPublish(exchange: exchangeBusiness,
                                             routingKey: routeKeyBusinessXiYiJiCommand,
                                             basicProperties: properties,
                                             body: body);
                    }

                }
            }
        }
    }
}