（2）RabbitMQ交換器Exchange介紹-有解無憂

1.AMQP Messaging中的基本概念

Broker：接收和分發訊息的應用，RabbitMQ Server就是Message Broker，
Virtual Host：出于多租戶和安全因素設計的，把AMQP的基本組件劃分到一個虛擬的分組中，類似于網路中的Namespace概念，當多個不同的用戶使用同一個RabbitMQ Server提供的服務時，可以劃分出多個Vhost，每個用戶在自己的Vhost創建Exchange／Queue等，
Connection：Producer、Consumer和Broker之間的TCP連接，斷開連接的操作只會在Client端進行，Broker不會斷開連接，除非出現網路故障或Broker服務出現問題，
Channel：如果每一次訪問RabbitMQ都建立一個Connection，在訊息量大的時候建立TCP Connection的開銷將是巨大的，效率也較低，Channel是在Connection內部建立的邏輯連接，如果應用程式支持多執行緒，通常每個Thread創建單獨的channel進行通訊，AMQP Method包含了Channel Id幫助客戶端和Message Broker識別Channel，所以Channel之間是完全隔離的，Channel作為輕量級的Connection極大減少了作業系統建立TCP Connection的開銷，
Exchange：Message到達Broker的第一站，根據分發規則，匹配查詢表中的Routing Key，分發訊息到Queue中去，常用的型別有：direct (point-to-point), topic (publish-subscribe) and fanout (multicast)，
Queue：訊息最終被送到這里等待Consumer取走，一個Message可以被同時拷貝到多個Queue中，
Binding：Exchange和Queue之間的虛擬連接，Binding中可以包含Routing Key，Binding資訊被保存到Exchange中的查詢表中，用于Message的分發依據，

2.Exchange交換機

很多時候我們都以為生產者是將訊息都投遞到Queue當中，實際上這在RabbitMQ中永遠不會發生，實際的情況是，生產者將訊息發送到Exchange（交換機），由Exchange將訊息路由到一個或多個Queue中或者丟棄，使用的路由演算法取決于交換機型別和其系結的規則，AMQP 0-9-1 Broker提供四種交換機型別：Direct exchange（直接交換機）、Fanout exchange（扇出交換機）、Topic exchange（話題交換機）、Headers exchange（標頭交換機），由于Headers exchange在實際使用中比較少，所以這里只重點介紹前三種模式，

2.1Direct exchange

由上述兩個流程圖可見，生產者所有發送到Direct exchange的訊息會通過指定Routing Key傳遞到Queue當中，

2.1 .1Default Exchange（默認交換機）

如果沒有沒有定義任何Exchange，默認就是Default Exchange，每個創建的Queue都使用與其名稱相同的Routing Key自動系結，具體示例，大家可以看看上一章節5.1，5.2小節，你會發現定義生產者代碼中QueueName與Routing Key都是同一個名稱，沒有定義任何Exchange系結（Binding）操作，只是在推送時候把Queue、Routing Key相同訊息推送出去，訊息傳遞時，Routing Key必須完全匹配，才會被Queue接收，否則該訊息會被拋棄，而Default Exchange也屬于Default Exchange模式，
下面來看看定義Direct exchange生產者的代碼：

class Program
{
    static void Main(string[] args)
    {
        string exchangeName = "DirectExchange";
        string queueName = "DirectExchangeQueueName";
        string routeKey = "DirectExchangeKey";

        //創建連接工廠
        var factory = new ConnectionFactory
        {
            UserName = "dengwu",//用戶名
            Password = "123456",//密碼
            HostName = "192.168.112.133",//rabbitmq ip
        };

        //創建連接
        var connection = factory.CreateConnection();
        //創建通道
        var channel = connection.CreateModel();

        //定義一個Direct型別交換機
        channel.ExchangeDeclare(exchangeName, ExchangeType.Direct, false, false, null);

        //定義一個佇列
        channel.QueueDeclare(queueName, false, false, false, null);

        //將佇列系結到交換機
        channel.QueueBind(queueName, exchangeName, routeKey, null);

        Console.WriteLine($"\nRabbitMQ連接成功,Exchange：{exchangeName}，Queue：{queueName}，Route：{routeKey}，\n\n請輸入訊息，輸入exit退出！");

        string input;
        do
        {
            input = Console.ReadLine();

            var sendBytes = Encoding.UTF8.GetBytes(input);
            //發布訊息
            channel.BasicPublish(exchangeName, routeKey, null, sendBytes);

        } while (input.Trim().ToLower() != "exit");
        channel.Close();
        connection.Close();
    }
}

運行：

2.2Fanout Exchange

由上述兩個流程圖可見，生產者所有發送到Fanout Exchange的訊息會轉發到與其系結(Binding)的所有Queue上，
Fanout Exchange模式無需通過指定Routing Key把訊息系結到Queue中，只需要把Exchange與Queue系結，這樣子通過Exchange訊息將會轉發到與其系結的所有Queue上，類似子網廣播，每臺子網內的主機都獲得了一份復制的訊息，
下面定義兩個佇列與其消費者FanoutExchangeQueueName1、FanoutExchangeQueueName2的Fanout Exchange生產者代碼：

class Program
{
    static void Main(string[] args)
    {
        string exchangeName = "FanoutExchange";
        string queueName1 = "FanoutExchangeQueueName1";
        string queueName2 = "FanoutExchangeQueueName2";
        string routeKey = "";

        //創建連接工廠
        var factory = new ConnectionFactory
        {
            UserName = "dengwu",//用戶名
            Password = "123456",//密碼
            HostName = "192.168.112.133",//rabbitmq ip
        };

        //創建連接
        var connection = factory.CreateConnection();
        //創建通道
        var channel = connection.CreateModel();

        //定義一個Fanout型別交換機
        channel.ExchangeDeclare(exchangeName, ExchangeType.Fanout, false, false, null);

        //定義佇列1
        channel.QueueDeclare(queueName1, false, false, false, null);
        //定義佇列2
        channel.QueueDeclare(queueName2, false, false, false, null);

        //將佇列系結到交換機
        channel.QueueBind(queueName1, exchangeName, routeKey, null);
        channel.QueueBind(queueName2, exchangeName, routeKey, null);

        //生成兩個佇列的消費者
        ConsumerGenerator(queueName1);
        ConsumerGenerator(queueName2);


        Console.WriteLine($"\nRabbitMQ連接成功，\n\n請輸入訊息，輸入exit退出！");

        string input;
        do
        {
            input = Console.ReadLine();

            var sendBytes = Encoding.UTF8.GetBytes(input);
            //發布訊息
            channel.BasicPublish(exchangeName, routeKey, null, sendBytes);

        } while (input.Trim().ToLower() != "exit");
        channel.Close();
        connection.Close();
    }

    /// <summary>
    /// 根據佇列名稱生成消費者
    /// </summary>
    /// <param name="queueName"></param>
    static void ConsumerGenerator(string queueName)
    {
        //創建連接工廠
        ConnectionFactory factory = new ConnectionFactory
        {
            UserName = "dengwu",//用戶名
            Password = "123456",//密碼
            HostName = "192.168.112.133",//rabbitmq ip
        };

        //創建連接
        var connection = factory.CreateConnection();
        //創建通道
        var channel = connection.CreateModel();

        //事件基本消費者
        var consumer = new EventingBasicConsumer(channel);

        //接收到訊息事件
        consumer.Received += (ch, ea) =>
        {
            var body = ea.Body.ToArray();
            var message = Encoding.UTF8.GetString(body);

            Console.WriteLine($"Queue:{queueName}收到訊息： {message}");
            //確認該訊息已被消費
            channel.BasicAck(ea.DeliveryTag, false);
        };
        //啟動消費者 設定為手動應答訊息
        channel.BasicConsume(queueName, false, consumer);
        Console.WriteLine($"Queue:{queueName}，消費者已啟動");
    }
}

運行：

2.3Topic Exchange

由上述流程圖可見，生產者所有發送到Topic Exchange的訊息會通過指定Routing Key被轉發到能和其匹配的Queue上，
Exchange指定Routing Key將路由進行模糊匹配，可以使用通配符進行模糊匹配，符號“#”匹配多個單詞（可以是零個），符號“*”匹配一個單詞，因此“usa.#”能夠匹配到“usa.news.xxx、usa.weather.xxx”，但是“usa.*” 只會匹配到“usa.news、usa.weather”，
所以Topic Exchange 使用是非常靈活的，Topic Exchange生產者代碼如下：

class Program
{
    static void Main(string[] args)
    {
        string exchangeName = "TopicExchange";
        string queueName = "DirectExchangeQueueName";
        string routeKey = "TopicExchangeKey.*";

        //創建連接工廠
        var factory = new ConnectionFactory
        {
            UserName = "dengwu",//用戶名
            Password = "123456",//密碼
            HostName = "192.168.112.133",//rabbitmq ip
        };

        //創建連接
        var connection = factory.CreateConnection();
        //創建通道
        var channel = connection.CreateModel();

        //定義一個Topic型別交換機
        channel.ExchangeDeclare(exchangeName, ExchangeType.Topic, false, false, null);

        //定義佇列1
        channel.QueueDeclare(queueName, false, false, false, null);

        //將佇列系結到交換機
        channel.QueueBind(queueName, exchangeName, routeKey, null);

        Console.WriteLine($"\nRabbitMQ連接成功，\n\n請輸入訊息，輸入exit退出！");

        string input;
        do
        {
            input = Console.ReadLine();

            var sendBytes = Encoding.UTF8.GetBytes(input);
            //發布訊息
            channel.BasicPublish(exchangeName, "TopicExchangeKey.one", null, sendBytes);

        } while (input.Trim().ToLower() != "exit");
        channel.Close();
        connection.Close();
    }
}

運行：

2.4Headers Exchange

Headers Exchange不依賴于Routing key 與 Binding key的匹配規則來路由訊息，而是根據發送的訊息內容中的 headers 屬性進行匹配，這種模式基本沒怎么用過，所以就跳過了，

參考文獻：
RabbitMQ官網
.NET Core 使用RabbitMQ

轉載請註明出處，本文鏈接：https://www.uj5u.com/net/4881.html

標籤：.NET Core

上一篇：（1）RabbitMQ簡介與安裝

下一篇：分享一個webapi介面性能測驗的工具