.netcore3.1 RabbitMq Header交换机

• .netcore3.1 RabbitMq Header交换机
  • 创建Header交换机与队列
  • 发布消息
  • 消费消息

其实这个玩意比较像我们在编程中的and与all的关系，简单来看可以如下面这张图所示。

创建Header交换机与队列

var factory = new ConnectionFactory()
{
    HostName = "47.98.187.188",
    UserName = "bob",
    Password = "bob"
};
// 创建一个链接
using (var connection = factory.CreateConnection())
{
    // 创建一个通道
    using (var channel = connection.CreateModel())
    {
        // 创建Header类型的交换机
        channel.ExchangeDeclare(exchange: "MyHeaderExchange", ExchangeType.Headers, true, false);
        // 声明一个队列
        channel.QueueDeclare(
            queue: "myheaderQueue1",
            durable: true,
            exclusive: false,
            autoDelete: false,
            arguments: null);
        // 绑定相关参数并将x-match设置为any
        channel.QueueBind("myheaderQueue1", "MyHeaderExchange",string.Empty,new Dictionary<string, object> {
            { "x-match","any" },
            { "name","aidasi" },
            { "age","18" }
        });
        Console.WriteLine("myheaderQueue1 Created！");
    }
}

var factory = new ConnectionFactory()
{
    HostName = "47.98.187.188",
    UserName = "bob",
    Password = "bob"
};
// 创建一个链接
using (var connection = factory.CreateConnection())
{
    // 创建一个通道
    using (var channel = connection.CreateModel())
    {
        // 创建Header类型的交换机
        channel.ExchangeDeclare(exchange: "MyHeaderExchange", ExchangeType.Headers, true, false);
        // 声明一个队列
        channel.QueueDeclare(
            queue: "myheaderQueue2",
            durable: true,
            exclusive: false,
            autoDelete: false,
            arguments: null);
        // 绑定相关参数并将x-match设置为any
        channel.QueueBind("myheaderQueue2", "MyHeaderExchange",string.Empty,new Dictionary<string, object> {
            { "x-match","all" },
            { "name","aidasi" },
            { "age","18" }
        });
        Console.WriteLine("myheaderQueue1 Created！");
    }
}

发布消息

var factory = new ConnectionFactory()
{
    HostName = "47.98.187.188",
    UserName = "bob",
    Password = "bob"
};
// 创建一个链接
using (var connection = factory.CreateConnection())
{
    // 创建一个通道
    using (var channel = connection.CreateModel())
    {
        var properties = channel.CreateBasicProperties();
        // 消息持久化
        properties.Persistent = true;
        // 实例化消息投
        // 创建any的队列
        properties.Headers = new Dictionary<string, object>() 
        {
            { "name","aidasi" },
            { "age","18" }
        };
        var msg = Encoding.UTF8.GetBytes("Message: name-aidasi,age-18");
        channel.BasicPublish("MyHeaderExchange", string.Empty, properties, msg);
        properties.Headers = new Dictionary<string, object>()
        {
            { "name","aidasi" }
        };
        var msg1 = Encoding.UTF8.GetBytes("Message: name-aidasi");
        channel.BasicPublish("MyHeaderExchange", string.Empty, properties, msg1);
        properties.Headers = new Dictionary<string, object>()
        {
            { "name","bob" },
            { "age","19" }
        };
        var msg2 = Encoding.UTF8.GetBytes("Message: name-bob,age-19");
        channel.BasicPublish("MyHeaderExchange", string.Empty, properties, msg2);
    }
}

消费消息

var factory = new ConnectionFactory()
{
    HostName = "47.98.187.188",
    UserName = "bob",
    Password = "bob"
};
Console.WriteLine("Please enter a queue name:");
// MyTopicExchange1 MyTopicExchange2 
string queuename = Console.ReadLine();
// 创建一个链接
using (var connection = factory.CreateConnection())
{
    // 创建一个通道
    using (var channel = connection.CreateModel())
    {
        // 创建消费实例
        var consumer = new EventingBasicConsumer(channel);
        // 事件在交付到使用者时触发。（消费处理事件）
        consumer.Received += (model, ea) =>
        {
            string EventQueueName = queuename;
            var body = ea.Body.ToArray();
            var message = Encoding.UTF8.GetString(body);
            Console.WriteLine($" Queue:{EventQueueName} Message: {message}");
            // 手动确认
            channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: true);
        };
        // 绑定到队列中去
        channel.BasicConsume(queue: queuename, autoAck: false, consumer: consumer);
        Console.ReadLine();
    }
}