.NET Core 使用RabbitMQ

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

RabbitMQ簡介

AMQP，即Advanced Message Queuing Protocol，高級消息隊列協議，是應用層協議的一個開放標準，爲面向消息的中間件設計。消息中間件主要用於組件之間的解耦，消息的發送者無需知道消息使用者的存在，反之亦然。算法

AMQP的主要特徵是面向消息、隊列、路由（包括點對點和發佈/訂閱）、可靠性、安全。
RabbitMQ是一個開源的AMQP實現，服務器端用Erlang語言編寫，支持多種客戶端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等，支持AJAX。用於在分佈式系統中存儲轉發消息，在易用性、擴展性、高可用性等方面表現不俗。centos

RabbitMQ提供了可靠的消息機制、跟蹤機制和靈活的消息路由，支持消息集羣和分佈式部署。適用於排隊算法、秒殺活動、消息分發、異步處理、數據同步、處理耗時任務、CQRS等應用場景。安全

RabbitMQ安裝

RabbitMQ安裝，網上已經有許多教程了，這裏簡單介紹一下在CentOS下安裝RabbitMQ。使用的版本爲3.6.12最新版。服務器

1.首先安裝erlang異步

rpm -Uvh https://www.rabbitmq.com/releases/erlang/erlang-19.0.4-1.el7.centos.x86_64.rpm分佈式

2.而後安裝socat學習

yum install socatui

3.最後安裝RabbitMQ.net

rpm -Uvh https://www.rabbitmq.com/releases/rabbitmq-server/v3.6.12/rabbitmq-server-3.6.12-1.el7.noarch.rpm設計

RabbitMQ經常使用命令

啓用Web控制檯

rabbitmq-plugins enable rabbitmq_management

開啓服務

systemctl start rabbitmq-server.service

中止服務

systemctl stop rabbitmq-server.service

查看服務狀態

systemctl status rabbitmq-server.service

查看RabbitMQ狀態

rabbitmqctl status

添加用戶賦予管理員權限

rabbitmqctl  add_user  username  password
rabbitmqctl  set_user_tags  username  administrator

查看用戶列表

rabbitmqctl list_users

刪除用戶

rabbitmqctl delete_user username

修改用戶密碼

rabbitmqctl oldPassword Username newPassword

訪問Web控制檯

http://服務器ip:15672/ 注意配置防火牆，默認用戶名密碼都是guest，若新建用戶必定要記得配置權限。

.NET Core 使用RabbitMQ

經過nuget安裝：https://www.nuget.org/packages/RabbitMQ.Client/

定義生產者
//建立鏈接工廠
ConnectionFactory factory = new ConnectionFactory
{
    UserName = "admin",//用戶名
    Password = "admin",//密碼
    HostName = "192.168.157.130"//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();
定義消費者
//建立鏈接工廠
            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纔會將這個消息從隊列中刪除，若是消費者在消費過程當中出現異常，斷開鏈接切沒有發送應答，那麼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}]");
};

演示：

從圖中能夠看出，設置了消息應答延遲10s，若是在這10s中，該消費者斷開了鏈接，那麼消息會被RabbitMQ從新投遞。

使用RabbitMQ的Exchange

前面咱們能夠看到生產者將消息投遞到Queue中，實際上這在RabbitMQ中這種事情永遠都不會發生。實際的狀況是，生產者將消息發送到Exchange（交換器），由Exchange將消息路由到一個或多個Queue中（或者丟棄）

AMQP協議中的核心思想就是生產者和消費者隔離，生產者從不直接將消息發送給隊列。生產者一般不知道是否一個消息會被髮送到隊列中，只是將消息發送到一個交換機。先由Exchange來接收，而後Exchange按照特定的策略轉發到Queue進行存儲。同理，消費者也是如此。Exchange 就相似於一個交換機，轉發各個消息分發到相應的隊列中。

RabbitMQ提供了四種Exchange模式：direct,fanout,topic,header 。可是 header模式在實際使用中較少，因此這裏只介紹前三種模式。

Exchange不是消費者關心的，因此消費者的代碼徹底不用變，用上面的消費者就好了。
因爲避免文章過長，影響閱讀，因此只貼了部分代碼，可是demo裏面是完整可運行的，詳細代碼請查看demo。

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

//定義隊列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, "TestRouteKey.one", null, sendBytes);

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

運行

Demo下載：DotNetCore.RabbitMQ

最後：歡迎加入 .net core 交流羣一塊兒學習，羣號：4656606 加入QQ羣

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