RabbitMQ與.net core(五) topic類型 與 headers類型 的Exchange

1.topic類型的Exchange

我們之前說過Topic類型的Exchange是direct類型的模糊查詢模式，可以通過routkey來實現模糊消費message，topic的模糊匹配有兩種模式：

1. 使用*來匹配一個單詞

2.使用#來匹配0個或多個單詞

我們來看代碼

消費端

using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Collections.Generic;
using System.Text;
using System.Threading;

namespace RabbitMQClient
{
    class Program
    {
        private static readonly ConnectionFactory rabbitMqFactory = new ConnectionFactory()
        {
            HostName = "39.**.**.**",
            Port = 5672,
            UserName = "root",
            Password = "root",
            VirtualHost = "/"
        };
        static void Main(string[] args)
        {
            var exchangeAll = "changeAll";
            var queueman = "queueman";
            var quemankey = "man.#";

            using (IConnection conn = rabbitMqFactory.CreateConnection())
            using (IModel channel = conn.CreateModel())
            {
                channel.ExchangeDeclare(exchangeAll, type: "topic", durable: true, autoDelete: false);
                channel.QueueDeclare(queueman, durable: true, exclusive: false, autoDelete: false);
                channel.QueueBind(queueman, exchangeAll, quemankey);

                channel.BasicQos(prefetchSize: 0, prefetchCount: 50, global: false);
                EventingBasicConsumer consumer = new EventingBasicConsumer(channel);
                consumer.Received += (model, ea) =>
                {
                    Byte[] body = ea.Body;
                    String message = Encoding.UTF8.GetString(body);
                    Console.WriteLine( message);
                    channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
                };

                channel.BasicConsume(queue: queueman, autoAck: false, consumer: consumer);
                Console.ReadLine();
            }
        }
    }
}

生產者代碼

using RabbitMQ.Client;
using System;
using System.Collections.Generic;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace RabbitMQConsole
{
    class Program
    {
        static void Main(string[] args)
        {
            ConnectionFactory factory = new ConnectionFactory();
            factory.HostName = "39.**.**.**";
            factory.Port = 5672;
            factory.VirtualHost = "/";
            factory.UserName = "root";
            factory.Password = "root";

            var exchangeAll = "changeAll";
            //性別.姓氏.頭髮長度
            var keymanA = "man.chen.long";
            var keymanB = "man.liu.long";
            var keymanC = "woman.liu.long";
            var keymanD = "woman.chen.short";

            using (var connection = factory.CreateConnection())
            {
                using (var channel = connection.CreateModel())
                {
                    channel.ExchangeDeclare(exchangeAll, type: "topic", durable: true, autoDelete: false);

                    var properties = channel.CreateBasicProperties();
                    properties.Persistent = true;
                    //發佈消息
                    channel.BasicPublish(exchange: exchangeAll,
                    routingKey: keymanA,
                    basicProperties: properties,
                    body: Encoding.UTF8.GetBytes(keymanA));
                    channel.BasicPublish(exchange: exchangeAll,
                     routingKey: keymanB,
                     basicProperties: properties,
                     body: Encoding.UTF8.GetBytes(keymanB));
                    channel.BasicPublish(exchange: exchangeAll,
                     routingKey: keymanC,
                     basicProperties: properties,
                     body: Encoding.UTF8.GetBytes(keymanC));
                    channel.BasicPublish(exchange: exchangeAll,
                     routingKey: keymanD,
                     basicProperties: properties,
                     body: Encoding.UTF8.GetBytes(keymanD));
                }
            }
        }
    }
}

我們先運行消費端再運行生產段，結果如下

消費端：

2.headers類型的exchange

生成者代碼

using RabbitMQ.Client;
using System;
using System.Collections.Generic;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace RabbitMQConsole
{
    class Program
    {
        static void Main(string[] args)
        {
            ConnectionFactory factory = new ConnectionFactory();
            factory.HostName = "39.**.**.**";
            factory.Port = 5672;
            factory.VirtualHost = "/";
            factory.UserName = "root";
            factory.Password = "root";

            var exchangeAll = "changeHeader";

            using (var connection = factory.CreateConnection())
            {
                using (var channel = connection.CreateModel())
                {
                    channel.ExchangeDeclare(exchangeAll, type: "headers", durable: true, autoDelete: false);

                    var properties = channel.CreateBasicProperties();
                    properties.Persistent = true;
                    properties.Headers = new Dictionary<string, object> {
                        { "sex","man"}
                    };
                    //發佈消息
                    channel.BasicPublish(exchange: exchangeAll,
                    routingKey: "",
                    basicProperties: properties,
                    body: Encoding.UTF8.GetBytes("hihihi"));
                }
            }
        }
    }
}

消費端代碼

using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Collections.Generic;
using System.Text;
using System.Threading;

namespace RabbitMQClient
{
    class Program
    {
        private static readonly ConnectionFactory rabbitMqFactory = new ConnectionFactory()
        {
            HostName = "39.**.**.**",
            Port = 5672,
            UserName = "root",
            Password = "root",
            VirtualHost = "/"
        };
        static void Main(string[] args)
        {
            var exchangeAll = "changeHeader";
            var queueman = "queueHeader";

            using (IConnection conn = rabbitMqFactory.CreateConnection())
            using (IModel channel = conn.CreateModel())
            {
                channel.ExchangeDeclare(exchangeAll, type: "headers", durable: true, autoDelete: false);
                channel.QueueDeclare(queueman, durable: true, exclusive: false, autoDelete: false);
                channel.QueueBind(queueman, exchangeAll, "",new Dictionary<string, object> { { "sex","man" } });

                channel.BasicQos(prefetchSize: 0, prefetchCount: 50, global: false);
                EventingBasicConsumer consumer = new EventingBasicConsumer(channel);
                consumer.Received += (model, ea) =>
                {
                    Byte[] body = ea.Body;
                    String message = Encoding.UTF8.GetString(body);
                    Console.WriteLine( message);
                    channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
                };

                channel.BasicConsume(queue: queueman, autoAck: false, consumer: consumer);
                Console.ReadLine();
            }
        }
    }
}