.NET Core使用RabbitMQ

時間  2019-12-17
標籤 core 使用 rabbitmq 欄目 RabbitMQ 简体版
原文   原文鏈接

RabbitMQ簡介

　RabbitMQ是一個開源的,基於AMQP(Advanced Message Queuing Protocol)協議的完整的可複用的企業級消息隊,RabbitMQ能夠實現點對點,發佈訂閱等消息處理模式。web

RabbitMQ是一個開源的AMQP實現,服務器端用Erlang語言編寫，支持Linux,windows,macOS,FreeBSD等操做系統,同時也支持不少語言,如：Python,Java,Ruby,PHP,C#,JavaScript,Go,Elixir,Objective-C,Swift等。ubuntu

RabbitMQ安裝

我使用的環境是ubuntu18.04,windows

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

消費者實現ui

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).gifspa

RabbitMQ的Worker模式

Worker模式其實就是一對多模式,咱們定義兩個消費者來看看效果操作系統

QQ20191210-104653-HD (1).gif

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

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

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程