.netcore3.1 RabbitMq Overflow,Dead Letter Exchange,Dead Letter Routing,Max priority

• .netcore3.1 RabbitMq Overflow,Dead Letter Exchange,Dead Letter Routing,Max priority
  • Overflow
    • 举个栗子
  • 死信(Dead Letter Exchange|Dead Letter Routing)
    • 死信的概念
    • 死信举例
    • 通过Demo实现举例
  • Max Priority（优先级）
    • Max Priority Demo

Overflow

举个栗子

var factory = new ConnectionFactory()
{
    HostName = "47.98.187.188",
    UserName = "bob",
    Password = "bob"
};
// 创建一个链接
using (var connection = factory.CreateConnection())
{
    // 创建一个通道
    using (var channel = connection.CreateModel())
    {
        // 声明一个只能为2条消息的队列，并将
        channel.QueueDeclare(
            queue: "Max_Length_Overflow_Queue",
            durable: true,
            exclusive: false,
            autoDelete: false,
            arguments: new Dictionary<string,object>{ 
                { "x-max-length",2 },
                { "x-overflow","reject-publish" }
            });
        var consumer = new EventingBasicConsumer(channel);
        // 绑定处理事件
        channel.BasicConsume(queue: "Max_Length_Overflow_Queue", autoAck: false, consumer: consumer);
        // 处理事件
        consumer.Received += (model, ea) =>
        {
            // 获取消息
            var body = ea.Body.ToArray();
            var props = ea.BasicProperties;
            try
            {
                // 人事做离职处理
                var message = Encoding.UTF8.GetString(body);
                Console.WriteLine("Tag:{0} Handle Message:{1}",ea.DeliveryTag, message);
            }
            catch (Exception e)
            {
                Console.WriteLine("Have Error:" + e.Message);
            }
            finally
            {
                channel.BasicAck(deliveryTag: ea.DeliveryTag,
                  multiple: true);
            }
        };
        // 发布8条消息
        for (int i = 0; i < 8; i++)
        {
            var msg = Encoding.UTF8.GetBytes("Good Morning!");
            channel.BasicPublish(string.Empty, "Max_Length_Overflow_Queue", null, msg);
        }
        Console.WriteLine(" Press [enter] to exit.");
        Console.ReadLine();
    }
}

死信(Dead Letter Exchange|Dead Letter Routing)

死信的概念

死信举例

相关参数

通过Demo实现举例

var factory = new ConnectionFactory()
{
    HostName = "47.98.187.188",
    UserName = "bob",
    Password = "bob"
};
// 创建一个链接
using (var connection = factory.CreateConnection())
{
    // 创建一个通道
    using (var channel = connection.CreateModel())
    {
        // 声明一个A餐厅的对立(位置就默认的老地方就可以了，我嫌麻烦)
        channel.QueueDeclare(
            queue: "A_Centeen_Queue",
            durable: false,
            exclusive: false,
            autoDelete: false,
            arguments: new Dictionary<string,object>{
                { "x-max-length",10 },
                { "x-dead-letter-exchange","B_Exchange" },
                { "x-dead-letter-routing-key","B_Centeen_Queue" },
            });
        // 发布11个人去吃饭的消息
        for (int i = 0; i < 12; i++)
        {
            var msg = Encoding.UTF8.GetBytes($"Person {i}");
            channel.BasicPublish(string.Empty, "A_Centeen_Queue", null, msg);
        }
        Console.WriteLine(" Press [enter] to exit.");
        Console.ReadLine();
    }
}

var factory = new ConnectionFactory()
{
    HostName = "47.98.187.188",
    UserName = "bob",
    Password = "bob"
};
// 创建一个链接
using (var connection = factory.CreateConnection())
{
    // 创建一个通道
    using (var channel = connection.CreateModel())
    {
        // 声明一个B餐厅位置的交换机
        channel.ExchangeDeclare("B_Exchange", ExchangeType.Direct, true, false, null);
        // 声明一个B餐厅
        channel.QueueDeclare(
            queue: "B_Centeen_Queue",
            durable: true,
            exclusive: false,
            autoDelete: false,
            arguments: null);
        // 绑定B餐厅
        channel.QueueBind("B_Centeen_Queue", "B_Exchange", "B_Centeen_Queue", null);
        // 我们这里只需要获取去B餐厅的那个人就可以了
        var consumer = new EventingBasicConsumer(channel);
        // 绑定处理事件
        channel.BasicConsume(queue: "B_Centeen_Queue", autoAck: false, consumer: consumer);
        // B餐厅消费
        consumer.Received += (model, ea) =>
        {
            var body = ea.Body.ToArray();
            var props = ea.BasicProperties;
            try
            {
                var message = Encoding.UTF8.GetString(body);
                Console.WriteLine("B Centeen Consumer Message: {0}", message);
            }
            catch (Exception e)
            {
                Console.WriteLine("Have Error:" + e.Message);
            }
            finally
            {
                channel.BasicAck(deliveryTag: ea.DeliveryTag,
                  multiple: false);
            }
        };
        Console.WriteLine(" Press [enter] to exit.");
        Console.ReadLine();
    }
}

Max Priority（优先级）

Max Priority Demo

var factory = new ConnectionFactory()
{
    HostName = "47.98.187.188",
    UserName = "bob",
    Password = "bob"
};
// 创建一个链接
using (var connection = factory.CreateConnection())
{
    // 创建一个通道
    using (var channel = connection.CreateModel())
    {
        // 声明一个Priority_Queue队列
        channel.QueueDeclare(
            queue: "Priority_Queue",
            durable: true,
            exclusive: false,
            autoDelete: false,
            arguments: new Dictionary<string,object>{ 
                // 设置优先级 0-10
                { "x-max-priority",10 },
            });
        // 发布Priority_Queue队列中的消息
        for (int i = 0; i < 10; i++)
        {
            var msg = Encoding.UTF8.GetBytes($"Message {i} ,Priority: 1");
            var cbp = channel.CreateBasicProperties();
            // 优先级都为1
            cbp.Priority = 1;
            channel.BasicPublish(string.Empty, "Priority_Queue", cbp, msg);
        }
        var cbphigh = channel.CreateBasicProperties();
        cbphigh.Priority = 10;
        // 发布一个高优先级消息
        channel.BasicPublish(string.Empty, "Priority_Queue", cbphigh, Encoding.UTF8.GetBytes($"High Priority Message,Priority: 10"));
        Console.WriteLine(" Press [enter] to exit.");
        Console.ReadLine();
    }
}