您好,登錄后才能下訂單哦!
這篇文章主要介紹“Rabbitmq如何使用”,在日常操作中,相信很多人在Rabbitmq如何使用問題上存在疑惑,小編查閱了各式資料,整理出簡單好用的操作方法,希望對大家解答”Rabbitmq如何使用”的疑惑有所幫助!接下來,請跟著小編一起來學習吧!
最簡單的使用
1.引用 rabbitMQ.Client
Install-Package RabbitMQ.Client
2.創建生產者
//創建連接工廠ConnectionFactory factory = new ConnectionFactory { UserName = "mqAdmin",//用戶名 Password = "mqAdmin",//密碼 HostName = "192.168.1.103"//rabbitmq ip};//創建連接var connection = factory.CreateConnection();//創建通道var channel = connection.CreateModel();//聲明一個隊列channel.QueueDeclare("hello", false, false, false, null); Console.WriteLine("\nRabbitMQ連接成功,請輸入消息,輸入exit退出!");string input;do{ input = Console.ReadLine(); var sendBytes = Encoding.UTF8.GetBytes(input); //發布消息 channel.BasicPublish("", "hello", null, sendBytes); } while (input.Trim().ToLower()!="exit"); channel.Close(); connection.Close();
3.創建生產者
//創建連接工廠ConnectionFactory factory = new ConnectionFactory { UserName = "admin",//用戶名 Password = "admin",//密碼 HostName = "192.168.157.130"//rabbitmq ip};//創建連接var connection = factory.CreateConnection();//創建通道var channel = connection.CreateModel();//事件基本消費者EventingBasicConsumer consumer = new EventingBasicConsumer(channel);//接收到消息事件consumer.Received += (ch, ea) =>{ var message = Encoding.UTF8.GetString(ea.Body); Console.WriteLine($"收到消息: {message}"); //確認該消息已被消費 channel.BasicAck(ea.DeliveryTag, false); };//啟動消費者 設置為手動應答消息channel.BasicConsume("hello", false, consumer); Console.WriteLine("消費者已啟動"); Console.ReadKey(); channel.Dispose(); connection.Close();
RabbitMQ消費失敗的處理
RabbitMQ采用消息應答機制,即消費者收到一個消息之后,需要發送一個應答,然后RabbitMQ才會將這個消息從隊列中刪除,如果消費者在消費過程中出現異常,斷開連接切沒有發送應答,那么RabbitMQ會將這個消息重新投遞。
//接收到消息事件consumer.Received += (ch, ea) => { var message = Encoding.UTF8.GetString(ea.Body); Console.WriteLine($"收到消息: {message}"); Console.WriteLine($"收到該消息[{ea.DeliveryTag}] 延遲10s發送回執"); Thread.Sleep(10000); //確認該消息已被消費 channel.BasicAck(ea.DeliveryTag, false); Console.WriteLine($"已發送回執[{ea.DeliveryTag}]"); };
使用RabbitMQ的Exchange
前面我們可以看到生產者將消息投遞到Queue中,實際上這在RabbitMQ中這種事情永遠都不會發生。實際的情況是,生產者將消息發送到Exchange(交換器),由Exchange將消息路由到一個或多個Queue中(或者丟棄)
AMQP協議中的核心思想就是生產者和消費者隔離,生產者從不直接將消息發送給隊列。生產者通常不知道是否一個消息會被發送到隊列中,只是將消息發送到一個交換機。先由Exchange來接收,然后Exchange按照特定的策略轉發到Queue進行存儲。同理,消費者也是如此。Exchange 就類似于一個交換機,轉發各個消息分發到相應的隊列中。
RabbitMQ提供了四種Exchange模式:direct,fanout,topic,header 。但是 header模式在實際使用中較少,所以這里只介紹前三種模式。
Exchange不是消費者關心的,所以消費者的代碼完全不用變,用上面的消費者就行了。
Direct Exchange
所有發送到Direct Exchange的消息被轉發到具有指定RouteKey的Queue。
Direct模式,可以使用rabbitMQ自帶的Exchange:default Exchange 。所以不需要將Exchange進行任何綁定(binding)操作 。消息傳遞時,RouteKey必須完全匹配,才會被隊列接收,否則該消息會被拋棄。
//創建連接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);
Fanout Exchange
所有發送到Fanout Exchange的消息都會被轉發到與該Exchange 綁定(Binding)的所有Queue上。
Fanout Exchange 不需要處理RouteKey 。只需要簡單的將隊列綁定到exchange 上。這樣發送到exchange的消息都會被轉發到與該交換機綁定的所有隊列上。類似子網廣播,每臺子網內的主機都獲得了一份復制的消息。
所以,Fanout Exchange 轉發消息是最快的。
為了演示效果,定義了兩個隊列,分別為hello1,hello2,每個隊列都擁有一個消費者。
static void Main(string[] args){ string exchangeName = "TestFanoutChange"; string queueName1 = "hello1"; string queueName2 = "hello2"; string routeKey = ""; //創建連接工廠 ConnectionFactory factory = new ConnectionFactory { UserName = "admin",//用戶名 Password = "admin",//密碼 HostName = "192.168.157.130"//rabbitmq ip }; //創建連接 var connection = factory.CreateConnection(); //創建通道 var channel = connection.CreateModel(); //定義一個Direct類型交換機 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 = "admin",//用戶名 Password = "admin",//密碼 HostName = "192.168.157.130"//rabbitmq ip }; //創建連接 var connection = factory.CreateConnection(); //創建通道 var channel = connection.CreateModel(); //事件基本消費者 EventingBasicConsumer consumer = new EventingBasicConsumer(channel); //接收到消息事件 consumer.Received += (ch, ea) => { var message = Encoding.UTF8.GetString(ea.Body); Console.WriteLine($"Queue:{queueName}收到消息: {message}"); //確認該消息已被消費 channel.BasicAck(ea.DeliveryTag, false); }; //啟動消費者 設置為手動應答消息 channel.BasicConsume(queueName, false, consumer); Console.WriteLine($"Queue:{queueName},消費者已啟動"); }
Topic Exchange
所有發送到Topic Exchange的消息被轉發到能和Topic匹配的Queue上, Exchange 將路由進行模糊匹配。可以使用通配符進行模糊匹配,符號“#”匹配一個或多個詞,符號“”匹配不多不少一個詞。因此“XiaoChen.#”能夠匹配到“XiaoChen.pets.cat”,但是“XiaoChen.” 只會匹配到“XiaoChen.money”。所以,Topic Exchange 使用非常靈活。string exchangeName = "TestTopicChange";string queueName = "hello";string routeKey = "TestRouteKey.*";//創建連接工廠ConnectionFactory factory = new ConnectionFactory { UserName = "admin",//用戶名 Password = "admin",//密碼 HostName = "192.168.157.130"//rabbitmq ip};//創建連接var connection = factory.CreateConnection();//創建通道var channel = connection.CreateModel();//定義一個Direct類型交換機channel.ExchangeDeclare(exchangeName, ExchangeType.Topic, false, false, null);//定義隊列1channel.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, "TestRouteKey.one", null, sendBytes); } while (input.Trim().ToLower() != "exit"); channel.Close(); connection.Close();
到此,關于“Rabbitmq如何使用”的學習就結束了,希望能夠解決大家的疑惑。理論與實踐的搭配能更好的幫助大家學習,快去試試吧!若想繼續學習更多相關知識,請繼續關注億速云網站,小編會繼續努力為大家帶來更多實用的文章!
免責聲明:本站發布的內容(圖片、視頻和文字)以原創、轉載和分享為主,文章觀點不代表本網站立場,如果涉及侵權請聯系站長郵箱:is@yisu.com進行舉報,并提供相關證據,一經查實,將立刻刪除涉嫌侵權內容。