碼上歡樂
相關內容    簡體    繁體

.NET Core使用RabbitMQ

本文轉載自   查看原文   2020-11-21 18:00   2145    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信息
    20191210093931.png

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(); } } }

同時運行兩個項目看效果
QQ20191210-102500-HD (1).gif

RabbitMQ的Worker模式

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

QQ20191210-104653-HD (1).gif

默認情況下，RabbitMQ會順序的將message發給下一個消費者。每個消費者會得到平均數量的message。這種方式稱之為round-robin（輪詢).
但是很多情況下並不希望消息平均分配,而是要消費快的多消費,消費少的少消費.還有很多情況下一旦其中一個宕機,那么另外接收者的無法接收原本這個接收者所要接收的數據。
我們修改其中一個消費者代碼,讓其等待3秒。在等待中停止運行 看看效果

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

20191210111323.png

當消費者宕機后消費者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);

QQ20191210-113454-HD (1).gif

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

‘能者多勞’模式

能者多勞是給消費速度快的消費更多的消息.少的責消費少的消息.能者多勞是建立在手動確認基礎上實現。
在延遲確認的消費中添加BasicQos

channel.QueueDeclare(
    queue: name,//消息隊列名稱 durable: false,//是否持久化,true持久化,隊列會保存磁盤,服務器重啟時可以保證不丟失相關信息。 exclusive: false,//是否排他,true排他的,如果一個隊列聲明為排他隊列,該隊列僅對首次聲明它的連接可見,並在連接斷開時自動刪除. autoDelete: false,//是否自動刪除。true是自動刪除。自動刪除的前提是：致少有一個消費者連接到這個隊列，之后所有與這個隊列連接的消費者都斷開時,才會自動刪除. arguments: null ////設置隊列的一些其它參數 ); //每次只能向消費者發送一條信息,再消費者未確認之前,不再向他發送信息 channel.BasicQos(0,1,false);

QQ20191210-120301-HD (1).gif

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

Exchange模式

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

發布訂閱模式(fanout)

20191210133305.png

生產者實現, 把隊列替換成了交換機, 發布消息時把交換機名稱告訴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(); } } }

QQ20191210-143430-HD (1).gif

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

路由模式(Direct Exchange)

20191210144538.png

路由模式下,在發布消息時指定不同的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(); } } }

20191210154431.png

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

20191210154636.png

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(); } } }

只有在通配符匹配通過的情況下才會接收消息,
20191210162012.png


免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。



猜您在找 .NET Core 使用RabbitMQ .NET Core 下使用 RabbitMQ RabbitMQ與.net core(一)安裝 .Net Core 3.1 WebAPI OR Mvc 中使用 RabbitMQ .NET Core微服務之基於EasyNetQ使用RabbitMQ消息隊列 RabbitMQ與.net core(二)Producer與Exchange .Net使用RabbitMQ詳解 NET 5 使用RabbitMQ以及Kafka區別 .Net core 使用TimeJob .NET core 使用Swagger
 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM