.NET Core使用RabbitMQ

RabbitMQ簡介

　RabbitMQ是一個開源的,基于AMQP(Advanced Message Queuing Protocol)協定的完整的可複用的企業級消息隊,RabbitMQ可以實作點對點,釋出訂閱等消息處理模式。

RabbitMQ是一個開源的AMQP實作,伺服器端用Erlang語言編寫，支援Linux,windows,macOS,FreeBSD等作業系統,同時也支援很多語言,如：Python,Java,Ruby,PHP,C#,JavaScript,Go,Elixir,Objective-C,Swift等。

RabbitMQ安裝

我使用的環境是ubuntu18.04,

RabbitMq需要Erlang語言的支援，在安裝RabbitMq之前需要安裝Erlang sudo apt-get install erlang-nox
更新源 sudo apt-get update
安裝RabbitMq sudo apt-get install rabbitmq-server
添加users使用者，密碼設定為admin sudo rabbitmqctl add_user users admin
給添加的使用者賦予權限 sudo rabbitmqctl set_user_tags users administrator
賦予virtual host中所有資源的配置、寫、讀權限以便管理其中的資源 rabbitmqctl set_permissions -p users '.*' '.*' '.*'
官方提供的一個web管理工具(rabbitmq_management),定位到Rabbitmq安裝目錄然後啟動web控制台 sudo rabbitmq-plugins enable rabbitmq_management
成功後可以在浏覽器輸入 http://localhost:15672/ 檢視RabbitMq資訊

.NET Core使用RabbitMQ

RabbitMQ常用指令

啟動 sudo rabbitmq-server start
停止 sudo rabbitmq-server stop
重新開機 sudo rabbitmq-server restart
檢視狀态 sudo rabbitmqctl status
檢視所有使用者 rabbitmqctl list_users
檢視使用者權限 rabbitmqctl list_user_permissions users
删除使用者權限 rabbitmqctl clear_permissions [-p VHostPath] users
删除使用者 rabbitmqctl delete_user users
修改使用者密碼 rabbitmqctl change_password users newpassword

.NET Core 使用RabbitMQ

通過 install-package rabbitmq.client 指令或nuget安裝 rabbitmq.client 包

生産者實作

using System;
using System.Text;
using RabbitMQ.Client;

namespace RabbitMQ
{
    class MainClass
    {
        static void Main(string[] args)
        {
            Console.WriteLine("生産者");
            IConnectionFactory factory = new ConnectionFactory//建立連接配接工廠對象
            {
                HostName = "106.12.90.208",//IP位址
                Port = 5672,//端口号
                UserName = "admin",//使用者賬号
                Password = "admin"//使用者密碼
            };
            IConnection con = factory.CreateConnection();//建立連接配接對象
            IModel channel = con.CreateModel();//建立連接配接會話對象
            string name = "demo";
            //聲明一個隊列
            channel.QueueDeclare(
              queue: name,//消息隊列名稱
              durable: false,//是否持久化,true持久化,隊列會儲存磁盤,伺服器重新開機時可以保證不丢失相關資訊。
              exclusive: false,//是否排他,true排他的,如果一個隊列聲明為排他隊列,該隊列僅對首次聲明它的連接配接可見,并在連接配接斷開時自動删除.
              autoDelete: false,//是否自動删除。true是自動删除。自動删除的前提是：緻少有一個消費者連接配接到這個隊列，之後所有與這個隊列連接配接的消費者都斷開時,才會自動删除.
              arguments: null //設定隊列的一些其它參數
               );
            string str = string.Empty;
            do
            {
                Console.WriteLine("發送内容:");
                str = Console.ReadLine();
                //消息内容
                byte[] body = Encoding.UTF8.GetBytes(str);
                //發送消息
                channel.BasicPublish("", name, null, body);
                Console.WriteLine("成功發送消息:" + str);
            }while(str.Trim().ToLower() != "exit");
            con.Close();
            channel.Close();
        }
    }
}

消費者實作

using System;
using System.Text;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;

namespace RabbitMQ
{
    class MainClass
    {

        static void Main(string[] args)
        {
            Console.WriteLine("消費者");
            IConnectionFactory factory = new ConnectionFactory//建立連接配接工廠對象
            {
                HostName = "106.12.90.208",//IP位址
                Port = 5672,//端口号
                UserName = "admin",//使用者賬号
                Password = "admin"//使用者密碼
            };
            IConnection conn = factory.CreateConnection();
            IModel channel = conn.CreateModel();
            string name = "demo";
            //聲明一個隊列
            channel.QueueDeclare(
              queue: name,//消息隊列名稱
              durable: false,//是否持久化,true持久化,隊列會儲存磁盤,伺服器重新開機時可以保證不丢失相關資訊。
              exclusive: false,//是否排他,true排他的,如果一個隊列聲明為排他隊列,該隊列僅對首次聲明它的連接配接可見,并在連接配接斷開時自動删除.
              autoDelete: false,//是否自動删除。true是自動删除。自動删除的前提是：緻少有一個消費者連接配接到這個隊列，之後所有與這個隊列連接配接的消費者都斷開時,才會自動删除.
              arguments: null ////設定隊列的一些其它參數
               );
            //建立消費者對象
            var consumer = new EventingBasicConsumer(channel);
            consumer.Received += (model, ea) =>
            {
                byte[] message = ea.Body;//接收到的消息
                        Console.WriteLine("接收到消息為:" + Encoding.UTF8.GetString(message));
            };
            //消費者開啟監聽
            channel.BasicConsume(name, true, consumer);
            Console.ReadKey();
            channel.Dispose();
            conn.Close();
        }

    }
}

同時運作兩個項目看效果

RabbitMQ的Worker模式

Worker模式其實就是一對多模式,我們定義兩個消費者來看看效果

預設情況下，RabbitMQ會順序的将message發給下一個消費者。每個消費者會得到平均數量的message。這種方式稱之為round-robin（輪詢).

但是很多情況下并不希望消息平均配置設定,而是要消費快的多消費,消費少的少消費.還有很多情況下一旦其中一個當機,那麼另外接收者的無法接收原本這個接收者所要接收的資料。

我們修改其中一個消費者代碼,讓其等待3秒。在等待中停止運作看看效果

consumer.Received += (model, ea) =>
{
    Thread.Sleep(3000);
    byte[] message = ea.Body;
    Console.WriteLine("接收到消息為:" + Encoding.UTF8.GetString(message));
};

當消費者當機後消費者1并沒有接受當機後的資料。是以我們需要消息确認來解決這個問題。

RabbitMQ消息确認

Rabbit中存在兩種消息确認模式

自動模式 - 隻要消息從隊列擷取,無論消費者擷取到消息後是否成功消費,都認為是消息成功消費.
手動模式 - 消費從隊列中擷取消息後,伺服器會将該消息處于不可用狀态,等待消費者回報。如果消費者在消費過程中出現異常，斷開連接配接切沒有發送應答，那麼RabbitMQ會将這個消息重新投遞。

修改兩個消費者代碼,并在其中一個中延遲确認。

consumer.Received += (model, ea) =>
{
    byte[] message = ea.Body;
    Console.WriteLine("接收到消息為:" + Encoding.UTF8.GetString(message));
    Thread.Sleep(3000); //等待三秒手動确認
    channel.BasicAck(ea.DeliveryTag, true);//傳回消息确認
};
////将autoAck設定false 關閉自動确認.
channel.BasicConsume(name, false, consumer);

如果在延遲中消費者斷開連接配接,那麼RabbitMQ會重新投遞未确認的消息

‘能者多勞’模式

能者多勞是給消費速度快的消費更多的消息.少的責消費少的消息.能者多勞是建立在手動确認基礎上實作。

在延遲确認的消費中添加BasicQos

channel.QueueDeclare(
    queue: name,//消息隊列名稱
    durable: false,//是否持久化,true持久化,隊列會儲存磁盤,伺服器重新開機時可以保證不丢失相關資訊。
    exclusive: false,//是否排他,true排他的,如果一個隊列聲明為排他隊列,該隊列僅對首次聲明它的連接配接可見,并在連接配接斷開時自動删除.
    autoDelete: false,//是否自動删除。true是自動删除。自動删除的前提是：緻少有一個消費者連接配接到這個隊列，之後所有與這個隊列連接配接的消費者都斷開時,才會自動删除.
    arguments: null ////設定隊列的一些其它參數
    );
//每次隻能向消費者發送一條資訊,再消費者未确認之前,不再向他發送資訊
channel.BasicQos(0,1,false);

可以看出消費快的消費者接受了更多的消息,這就是能者多勞模式的展現

Exchange模式

在RabbitMQ的Exchange模式中生産者并不會直接把消息發送到Queue中,而是将消息發送到Exchange（交換機),消費者建立各自的隊列綁定到交換機.

釋出訂閱模式(fanout)

生産者實作, 把隊列替換成了交換機, 釋出消息時把交換機名稱告訴RabbitMQ,把交換機設定成fanout釋出訂閱模式

using System;
using System.Text;
using RabbitMQ.Client;

namespace RabbitMQ
{
    class MainClass
    {
        static void Main(string[] args)
        {
            Console.WriteLine("生産者");
            IConnectionFactory factory = new ConnectionFactory//建立連接配接工廠對象
            {
                HostName = "106.12.90.208",//IP位址
                Port = 5672,//端口号
                UserName = "admin",//使用者賬号
                Password = "admin"//使用者密碼
            };
            IConnection con = factory.CreateConnection();//建立連接配接對象
            IModel channel = con.CreateModel();//建立連接配接會話對象
            sstring exchangeName = "exchange1";//交換機名稱
            //把交換機設定成fanout釋出訂閱模式
            channel.ExchangeDeclare(name, type: "fanout");
            string str;
            do
            {
                str = Console.ReadLine();
                //消息内容
                byte[] body = Encoding.UTF8.GetBytes(str);
                //釋出消息,
                channel.BasicPublish(exchangeName, "", null, body);
            }while(str.Trim().ToLower() != "exit");
            con.Close();
            channel.Close();
        }
    }
}

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

namespace mq
{
    class MainClass
    {

        static void Main(string[] args)
        {
            IConnectionFactory factory = new ConnectionFactory//建立連接配接工廠對象
            {
                HostName = "106.12.90.208",//IP位址
                Port = 5672,//端口号
                UserName = "admin",//使用者賬号
                Password = "admin"//使用者密碼
            };
            IConnection conn = factory.CreateConnection();
            IModel channel = conn.CreateModel();
            //交換機名稱
            string exchangeName = "exchange1";
            //聲明交換機
            channel.ExchangeDeclare(exchangeName, ExchangeType.Fanout);
            //消息隊列名稱
            string queueName = DateTime.Now.Second.ToString();
            //聲明隊列
            channel.QueueDeclare(queueName, false, false, false, null);
            //将隊列與交換機進行綁定
            channel.QueueBind(queueName, exchangeName, "", null);
            //定義消費者
            var consumer = new EventingBasicConsumer(channel);
            Console.WriteLine($"隊列名稱:{queueName}");
            //接收事件
            consumer.Received += (model, ea) =>
            {
                byte[] message = ea.Body;//接收到的消息
                Console.WriteLine($"接收到資訊為:{Encoding.UTF8.GetString(message)}");
                //傳回消息确認
                channel.BasicAck(ea.DeliveryTag, true);
            };
            //開啟監聽
            channel.BasicConsume(queueName, false, consumer);
            Console.ReadKey();
        }
    }
}

當消費者綁定同樣的交換機,可以看到兩個不同的消費者都能接受到生産者發送的所有消息。

路由模式(Direct Exchange)

路由模式下,在釋出消息時指定不同的routeKey,交換機會根據不同的routeKey分發消息到不同的隊列中

using System;
using System.Text;
using RabbitMQ.Client;

namespace RabbitMQ
{
    class MainClass
    {
        static void Main(string[] args)
        {
            Console.WriteLine("生産者");
            IConnectionFactory factory = new ConnectionFactory//建立連接配接工廠對象
            {
                HostName = "106.12.90.208",//IP位址
                Port = 5672,//端口号
                UserName = "admin",//使用者賬号
                Password = "admin"//使用者密碼
            };
            IConnection con = factory.CreateConnection();//建立連接配接對象
            IModel channel = con.CreateModel();//建立連接配接會話對象
            string exchangeName = "exchange1"; //交換機名稱
            string routeKey = "key1"; //比對的key，
            //把交換機設定成Direct模式
            channel.ExchangeDeclare(exchangeName, ExchangeType.Direct);
            string str;
            do
            {
                str = Console.ReadLine();
                //消息内容
                byte[] body = Encoding.UTF8.GetBytes(str);
                //發送消息
                channel.BasicPublish(exchangeName, routeKey, null, body);
            }while(str.Trim().ToLower() != "exit");
            con.Close();
            channel.Close();
        }

    }
}

申明一個routeKey值為key1,并在釋出消息的時候告訴了RabbitMQ,消息傳遞時routeKey必須比對,才會被隊列接收否則消息會被抛棄。

using System;
using System.Text;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;

namespace mq
{
    class MainClass
    {

        static void Main(string[] args)
        {
            Console.WriteLine($"輸入接受key名稱:");
            string routeKey = Console.ReadLine();
            IConnectionFactory factory = new ConnectionFactory//建立連接配接工廠對象
            {
                HostName = "106.12.90.208",//IP位址
                Port = 5672,//端口号
                UserName = "admin",//使用者賬号
                Password = "admin"//使用者密碼
            };
            IConnection conn = factory.CreateConnection();
            IModel channel = conn.CreateModel();
            //交換機名稱
            string exchangeName = "exchange11";
            //聲明交換機
            channel.ExchangeDeclare(exchangeName, ExchangeType.Direct);
            //消息隊列名稱
            string queueName = DateTime.Now.Second.ToString();
            //聲明隊列
            channel.QueueDeclare(queueName, false, false, false, null);
            //将隊列,key與交換機進行綁定
            channel.QueueBind(queueName, exchangeName, routeKey, null);
            //定義消費者
            var consumer = new EventingBasicConsumer(channel);
            Console.WriteLine($"隊列名稱:{queueName}");
            //接收事件
            consumer.Received += (model, ea) =>
            {
                byte[] message = ea.Body;//接收到的消息
                Console.WriteLine($"接收到資訊為:{Encoding.UTF8.GetString(message)}");
                //傳回消息确認
                channel.BasicAck(ea.DeliveryTag, true);
            };
            //開啟監聽
            channel.BasicConsume(queueName, false, consumer);
            Console.ReadKey();
        }
    }

}

在routeKey比對的時候才會接收消息,接收者消息隊列可以聲明多個routeKey與交換機進行綁定

routeKey不比對則不接收消息。

通配符模式(Topic Exchange)

通配符模式和路由模式其實差不多,不同于配符模式中的路由可以聲明為模糊查詢.

符号“#”比對一個或多個詞.

符号“*”比對一個詞。

RabbitMQ中通配符的通配符是用"."來分割字元串的.比如a.*隻能比對到a.b,a.c,而a.#可以比對到a.a.c,a.a.b.

生成者實作

using System;
using System.Text;
using RabbitMQ.Client;

namespace RabbitMQ
{
    class MainClass
    {
        static void Main(string[] args)
        {
            Console.WriteLine("生産者");
            IConnectionFactory factory = new ConnectionFactory//建立連接配接工廠對象
            {
                HostName = "106.12.90.208",//IP位址
                Port = 5672,//端口号
                UserName = "admin",//使用者賬号
                Password = "admin"//使用者密碼
            };
            IConnection con = factory.CreateConnection();//建立連接配接對象
            IModel channel = con.CreateModel();//建立連接配接會話對象
            string exchangeName = "exchange114"; //交換機名稱
            string routeKey = "key.a"; //比對的key，
            //把交換機設定成Topic模式
            channel.ExchangeDeclare(exchangeName, ExchangeType.Topic);
            string str;
            do
            {
                str = Console.ReadLine();
                //消息内容
                byte[] body = Encoding.UTF8.GetBytes(str);
                //發送消息
                channel.BasicPublish(exchangeName, routeKey, null, body);
            }while(str.Trim().ToLower() != "exit");
            con.Close();
            channel.Close();
        }

    }
}

using System;
using System.Text;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;

namespace mq
{
    class MainClass
    {

        static void Main(string[] args)
        {
            Console.WriteLine($"輸入接受key名稱:");
            string routeKey = "key.*"; //使用通配符來比對key
            IConnectionFactory factory = new ConnectionFactory//建立連接配接工廠對象
            {
                HostName = "106.12.90.208",//IP位址
                Port = 5672,//端口号
                UserName = "admin",//使用者賬号
                Password = "admin"//使用者密碼
            };
            IConnection conn = factory.CreateConnection();
            IModel channel = conn.CreateModel();
            //交換機名稱
            string exchangeName = "exchange114";
            //聲明交換機
            channel.ExchangeDeclare(exchangeName, ExchangeType.Topic);
            //消息隊列名稱
            string queueName = DateTime.Now.Second.ToString();
            //聲明隊列
            channel.QueueDeclare(queueName, false, false, false, null);
            //将隊列與交換機進行綁定
            channel.QueueBind(queueName, exchangeName, routeKey, null);
            //定義消費者
            var consumer = new EventingBasicConsumer(channel);
            Console.WriteLine($"隊列名稱:{queueName}");
            //接收事件
            consumer.Received += (model, ea) =>
            {
                byte[] message = ea.Body;//接收到的消息
                Console.WriteLine($"接收到資訊為:{Encoding.UTF8.GetString(message)}");
                //傳回消息确認
                channel.BasicAck(ea.DeliveryTag, true);
            };
            //開啟監聽
            channel.BasicConsume(queueName, false, consumer);
            Console.ReadKey();
        }
    }

}

隻有在通配符比對通過的情況下才會接收消息,