.netcore3.1 RabbitMq Single Active Consumer

• .netcore3.1 RabbitMq Single Active Consumer
  • Single Active Consumer
    • 创建队列
    • 创建消费端代码
    • 添加消息
    • 调整消费端处理量

Single Active Consumer

创建队列

static void Main(string[] args)
{
    var factory = new ConnectionFactory()
    {
        HostName = "47.98.187.188",
        UserName = "bob",
        Password = "bob"
    };
    // 创建一个链接
    using (var connection = factory.CreateConnection())
    {
        // 创建一个通道
        using (var channel = connection.CreateModel())
        {
            // 声明一个Single_Active_Consumer_Queue队列
            channel.QueueDeclare(
                queue: "Single_Active_Consumer_Queue",
                durable: true,
                exclusive: false,
                autoDelete: false,
                arguments: new Dictionary<string,object>{
                    { "x-single-active-consumer",true },
                });
            Console.WriteLine(" Press [enter] to exit.");
            Console.ReadLine();
        }
    }
}

创建消费端代码

var factory = new ConnectionFactory()
{
    HostName = "47.98.187.188",
    UserName = "bob",
    Password = "bob"
};
string Name = args[0];
// 创建一个链接
using (var connection = factory.CreateConnection())
{
    // 创建一个通道
    using (var channel = connection.CreateModel())
    {
        // 我们这里只需要获取去B餐厅的那个人就可以了
        var consumer = new EventingBasicConsumer(channel);
        // 绑定处理事件
        channel.BasicConsume(queue: "Single_Active_Consumer_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("{0} Message: {1}", Name, 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();
    }
}

添加消息

for (int i = 0; i < 10; i++)
{
    var msg = Encoding.UTF8.GetBytes($"Message {i} ,Priority: 1");
    channel.BasicPublish(string.Empty, "Single_Active_Consumer_Queue", null, msg);
}

调整消费端处理量

// 我们这里只需要获取去B餐厅的那个人就可以了
var consumer = new EventingBasicConsumer(channel);
// 绑定处理事件
channel.BasicConsume(queue: "Single_Active_Consumer_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("{0} Message: {1}", Name, message);
        // 处理4条就退出
        if (Name.Contains("A") && ea.DeliveryTag == 4)
        {
            Console.WriteLine("A go home!!!!");
            // 退出程序的代码
            Environment.Exit(0);
        }
    }
    catch (Exception e)
    {
        Console.WriteLine("Have Error:" + e.Message);
    }
    finally
    {
        channel.BasicAck(deliveryTag: ea.DeliveryTag,
          multiple: false);
    }
};
Console.WriteLine(" Press [enter] to exit.");

channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);