Rabbitmq之死信队列

在使用rabbitmq过程中可能会遇到的情况:

    1、消费者挂了,导致队列积压,超出最大长度,超时若有新的消息过来,这消息会被丢弃或发送不成功;
    2、对于特殊场景的消息,可能有需要处理消息超时的需求;

以上两种情况,应用rabbitmq的死信队列可以比较好的解决。死信队列也是一个正常的exchange,其下绑定了一个或多个queue。这个exchange有dead-letter-exchange的标志。
而消息进入死信队列的情况主要有
1. 消息设置了ttl,并且超时;
2. 消息被拒绝;
3. 队列未被消费的消息数量达到了队列的最大长度。

废话不多说,下面做个简单的demo,直接上代码(.net core3.1 + Rabbitmq.Client 6.1):

1、DMRabbitmqClient.cs, 简单封装的rabbitmqclient。

    public class DMRabbitmqClient
    {
        private ConnectionFactory Factory { get; }

        public DMRabbitmqClient(string host, string username = "guest", string password = "guest", int port = 5672)
        {
            Factory = new ConnectionFactory();
            Factory.UserName = username;
            Factory.HostName = host;
            Factory.Password = password;
            Factory.Port = port;
        }

        public IConnection CreateConnection()
        {
            return Factory.CreateConnection();
        }

        public IModel GetModel(IConnection connection)
        {
            return connection.CreateModel();
        }

        public void ExchangeDeclare(IModel model, string exchangename, string type, bool durable, bool autoDelete, IDictionary<string, object> arguments)
        {
            if (arguments == null)
                arguments = new Dictionary<string, object>();
            model.ExchangeDeclare(exchangename, type, durable, autoDelete, arguments);
        }

        public void QueueDeclare(IModel model, string queue, bool durable, bool exclusive, bool autoDelete, IDictionary<string, object> arguments)
        {
            if (arguments == null)
                arguments = new Dictionary<string, object>();
            model.QueueDeclare(queue, durable, exclusive, autoDelete, arguments);
        }

        public void QueueBind(IModel model, string queue, string exchange, string routeKey, IDictionary<string, object> arguments)
        {
            if (arguments == null)
                arguments = new Dictionary<string, object>();
            model.QueueBind(queue, exchange, routeKey, arguments);
        }

        public void Publish(IModel model, string dto, string exchange, string routekey)
        {
            IBasicProperties properties = model.CreateBasicProperties();
            properties.DeliveryMode = 2;
            properties.Expiration = "10000";
            properties.ContentEncoding = "UTF-8";

            var content = Encoding.UTF8.GetBytes(dto);

            model.BasicPublish(exchange, routekey, false, properties, content);
        }

        public EventingBasicConsumer AddConsumerEvent(IModel model,EventHandler<BasicDeliverEventArgs> action)
        {
            var eventConsumer = new EventingBasicConsumer(model);
            eventConsumer.Received += action;
            return eventConsumer;
        }

    }

2、生产者代码

```
class Program
{
    static void Main(string[] args)
    {
        var client = new DMRabbitmqClient("127.0.0.1", "guest", "guest");
        IConnection connection = client.CreateConnection();
        var model = client.GetModel(connection);

        //定义死信队列
        client.ExchangeDeclare(model, "test_exchange_dl", "topic", false, false, null);
        client.QueueDeclare(model, "test_queue_dl", false, false, false, null);
        //exchange接收到消息都往这个队列里面发
        client.QueueBind(model, "test_queue_dl", "test_exchange_dl", "#", null);
        //定义死信队列消费者
        var dlconsumer = client.AddConsumerEvent(model,DLReceiver);
        model.BasicConsume("test_queue_dl", true, "", false, true, null, dlconsumer);


        //定义正常的队列
        IModel common_model = client.GetModel(connection);
        client.ExchangeDeclare(common_model, "test_exchange", "topic", false, false, null);
        var arguments = new Dictionary<string, object>();
        arguments.Add("x-dead-letter-exchange", "test_exchange_dl");//定义reject或者ttl超时后,发送到死信队列的exchange
        arguments.Add("x-max-length", 3);//定义队列最大长度
        client.QueueDeclare(common_model, "test_queue", false, false, false, arguments);
        client.QueueBind(common_model, "test_queue", "test_exchange", "test_#", null);

        Console.WriteLine("输入1、发送消息;2、退出");
        string key = Console.ReadLine();
        bool isExists = false;
        while (!string.IsNullOrWhiteSpace(key))
        {
            switch (key)
            {
                case "1":
                    client.Publish(common_model, $"{DateTime.Now:yyyy-MM-dd HH:mm:ss},您好。", "test_exchange", "test_#");
                    break;
                case "2":
                    isExists = true;
                    break;
            }
            if (isExists)
                break;
            Console.WriteLine("输入1、发送消息;2、退出");
            key = Console.ReadLine();
        }

        Console.WriteLine("please pass any key to exists");
        Console.ReadKey();
    }

    private static void DLReceiver(object sender, BasicDeliverEventArgs args)
    {
        string msg = Encoding.UTF8.GetString(args.Body.ToArray());

        Console.WriteLine($"{DateTime.Now:yyyy-MM-dd HH:mm:ss}死信队列接收到消息:{msg}");

    }
}

2、消费者代码

    ```
    class Program
    {
        static void Main(string[] args)
        {
            var client = new DMRabbitmqClient("127.0.0.1", "guest", "guest");
            IConnection connection = client.CreateConnection();
            var model = client.GetModel(connection);
            //model.ExchangeDeclare("test_exchange", "topic", false, false, null);
            //model.QueueDeclare("test_queue", true, true, false);

            Console.WriteLine("请输入:1、拉取消息并自动ack;2、拉取消息但是不ack;3、退出。");
            string key = Console.ReadLine();
            bool isExists = false;
            while (!string.IsNullOrWhiteSpace(key))
            {
                switch (key)
                {
                    case "1":
                        PullMessage(model, true);
                        break;
                    case "2":
                        PullMessage(model, false);
                        break;
                    case "3":
                        isExists = true;
                        break;
                }
                if (isExists)
                    break;
                Console.WriteLine("请输入:1、拉取消息并自动ack;2、拉取消息但是不ack;3、退出。");
                key = Console.ReadLine();
            }



            Console.WriteLine("please pass any keys to exists");
            Console.ReadKey();
        }

        private static void PullMessage(IModel model ,bool isAck)
        {
            var result = model.BasicGet("test_queue", isAck);
            if (result != null && !result.Body.IsEmpty)
            {
                string message = Encoding.UTF8.GetString(result.Body.ToArray());
                Console.WriteLine($"{DateTime.Now:yyyy-MM-dd HH:mm:ss}接收到消息:{message},ack:{isAck}");
                if (!isAck)
                {
                    Console.WriteLine("ack为false,拒绝...");
                    model.BasicReject(1, false);//拒绝ack,并且deliverytag设置为1,requeue设置为false
                }
            }
        }
    }
  实践
  1、生产者发送消息,但是不主动拉取消息,此时消息进入私信队列。如下图

Rabbitmq之死信队列

  2、生产者发送消息,消费者端接收消息,但是调用basereject方法拒绝。如下图

Rabbitmq之死信队列

  3、此处生产者代码已经设置了最大队列长度为3,我们尝试连续发送消息,让队列积压的消息达到最大长度,如下图

Rabbitmq之死信队列

上一篇:命名路由传参


下一篇:11-13