RabbitMQ系列教程之七：RabbitMQ的 C# 客戶端 API 的簡(jiǎn)介

發(fā)布時(shí)間：2017年07月18日作者： IT網(wǎng)絡(luò)文摘 (該文來(lái)自筆記，點(diǎn)擊查看原文)

今天這篇博文是我翻譯的RabbitMQ的最后一篇文章了，介紹一下RabbitMQ的C#開發(fā)的接口。好了，言歸正傳吧。

   Net/C# 客戶端 API簡(jiǎn)介

1、主要的命名空間，接口和類

定義核心的API的接口和類被定義在RabbitMQ.Client這個(gè)命名空間下面：

所以要想使用RabbitMQ的功能，需要以下代碼

   using RabbitMQ.Client;

   【1】、核心API的接口和類如下：

    IModel:表示一個(gè)符合AMQP 0-9-1 協(xié)議的通道，并且提供了很多的操作方法

    IConnection:表示一個(gè)符合AMQP 0-9-1協(xié)議的連接對(duì)象，用戶和RabbitMQ 服務(wù)端的連接

    ConnectionFactory:可以創(chuàng)建一個(gè)IConnection對(duì)象的實(shí)例。

    IBasicConsumer:表示一個(gè)消息的消費(fèi)者，或者是使用者。

   【2】、其他有用的接口和類包含如下：

    DefaultBasicConsumer:通常用作消費(fèi)者的基類，如果要編寫自己的消費(fèi)者程序，可以從該類繼承。

   【3】、除RabbitMQ.Client之外的公共命名空間包括：

    RabbitMQ.Client.Events:作為客戶端庫(kù)一部分的各種事件和事件處理程序。

       包括EventingBasicConsumer，一個(gè)基于C＃事件處理程序構(gòu)建的消費(fèi)者實(shí)現(xiàn)。

    RabbitMQ.Client.Exceptions: 對(duì)用戶可見的一些異常對(duì)象。

    所有其他命名空間都保留用于庫(kù)的私有實(shí)現(xiàn)細(xì)節(jié)，盡管私有命名空間的成員通?？梢允褂迷搸?kù)的應(yīng)用程序使用，以便允許開發(fā)人員實(shí)現(xiàn)其在庫(kù)實(shí)現(xiàn)中發(fā)現(xiàn)的故障或設(shè)計(jì)錯(cuò)誤的解決方法。應(yīng)用程序不能依賴于在庫(kù)的版本中保持穩(wěn)定的私有命名空間中出現(xiàn)的任何類，接口，成員變量等。

2、創(chuàng)建到代理的連接

要連接到RabbitMQ，需要實(shí)例化一個(gè)ConnectionFactory并將其配置為使用所需的主機(jī)名，虛擬主機(jī)和憑據(jù)。然后使用ConnectionFactory.CreateConnection（）打開一個(gè)連接。以下兩個(gè)代碼片段連接到hostName上的RabbitMQ節(jié)點(diǎn)：

ConnectionFactory factory = new ConnectionFactory();   // "guest"/"guest" by default, limited to localhost connections
   factory.UserName = user;
   factory.Password = pass;
   factory.VirtualHost = vhost;
   factory.HostName = hostName;
   IConnection conn = factory.CreateConnection();


   ConnectionFactory factory = new ConnectionFactory();
   factory.Uri = "amqp://user:pass@hostName:port/vhost";
   IConnection conn = factory.CreateConnection();

由于.NET客戶端使用比其他客戶端更嚴(yán)格的AMQP 0-9-1 URI規(guī)范解釋，因此在使用URI時(shí)必須小心。特別是，主機(jī)部分不能被忽略，具有空名稱的虛擬主機(jī)不可尋址。所有出廠屬性都有默認(rèn)值。如果屬性在創(chuàng)建連接之前保持未分配，則將使用屬性的默認(rèn)值：

Username    "guest"
   Password    "guest"
   Virtual host    "/"
   Hostname    "localhost"
   port    5672 是針對(duì)一般而言的連接, 5671 是針對(duì) TLS 連接的

IConnection接口可以打開一個(gè)通道:

 IModel channel = conn.CreateModel();

通道Channel用于接收和發(fā)送消息

3、使用消息交換機(jī)和隊(duì)列

客戶端應(yīng)用程序?qū)⑴c消息交換機(jī)和消息隊(duì)列（AMQP 0-9-1的高級(jí)構(gòu)建塊）配合工作。【消息交換機(jī)】和【消息隊(duì)列】在使用之前必須先聲明他們。聲明任何類型的對(duì)象只是確保其中一個(gè)名稱存在，如有必要，創(chuàng)建它。繼續(xù)前面的例子，以下代碼聲明一個(gè)消息交換機(jī)和一個(gè)隊(duì)列，然后將它們綁定在一起。

model.ExchangeDeclare(exchangeName, ExchangeType.Direct);
   model.QueueDeclare(queueName, false, false, false, null);
   model.QueueBind(queueName, exchangeName, routingKey, null);

這將聲明了以下兩個(gè)對(duì)象：

【1】、非持久、非自動(dòng)刪除的、交換類型為“direct”的消息交換機(jī);

【2】、非持久、非自動(dòng)刪除、非排他的消息隊(duì)列

可以通過(guò)使用附加參數(shù)來(lái)定制消息交換機(jī)。然后，上面的代碼使用給定的路由鍵將隊(duì)列綁定到消息交換機(jī)。請(qǐng)注意，許多通道API（IModel）方法重載。 ExchangeDeclare方便的簡(jiǎn)單形式使用合理的默認(rèn)值。還有更多的表單具有更多的參數(shù)，可以根據(jù)需要修改這些默認(rèn)值，并在需要時(shí)進(jìn)行完全控制。在整個(gè)API中使用這種“短版本，長(zhǎng)版本”模式。

4、發(fā)布消息

要將消息發(fā)布到消息交換機(jī)，請(qǐng)使用IModel.BasicPublish，如下所示：

 byte[] messageBodyBytes = System.Text.Encoding.UTF8.GetBytes("Hello, world!");
  model.BasicPublish(exchangeName, routingKey, null, messageBodyBytes);

為了精細(xì)控制，您可以使用重載變量來(lái)指定強(qiáng)制標(biāo)志，或指定消息屬性：

byte[] messageBodyBytes = System.Text.Encoding.UTF8.GetBytes("Hello, world!");
  IBasicProperties props = model.CreateBasicProperties();
  props.ContentType = "text/plain";
  props.DeliveryMode = 2;
  model.BasicPublish(exchangeName,
                   routingKey, props,
                   messageBodyBytes);

這將發(fā)送一個(gè)帶有發(fā)送模式為2（持久性）和內(nèi)容類型是“text / plain”的消息。有關(guān)可用消息屬性的更多信息，請(qǐng)參閱IBasicProperties接口的定義。

在以下示例中，我們使用自定義標(biāo)頭發(fā)布消息：

byte[] messageBodyBytes = System.Text.Encoding.UTF8.GetBytes("Hello, world!");

   IBasicProperties props = model.CreateBasicProperties();
   props.ContentType = "text/plain";
   props.DeliveryMode = 2;
   props.Headers = new Dictionary<string, object>();
   props.Headers.Add("latitude",  51.5252949);
   props.Headers.Add("longitude", -0.0905493);

   model.BasicPublish(exchangeName,
                   routingKey, props,
                   messageBodyBytes);

下面的示例設(shè)置消息過(guò)期：

byte[] messageBodyBytes = System.Text.Encoding.UTF8.GetBytes("Hello, world!");

   IBasicProperties props = model.CreateBasicProperties();
   props.ContentType = "text/plain";
   props.DeliveryMode = 2;
   props.Expiration = "36000000"

   mode.BasicPublish(exchangeName,
                  routingKey, props,
                  messageBodyBytes);

5、獲取個(gè)人消息（“拉API”）

要檢索單個(gè)消息，請(qǐng)使用IModel.BasicGet。返回的值是BasicGetResult的實(shí)例，可以從中提取基本屬性和消息體：

bool noAck = false;
BasicGetResult result = channel.BasicGet(queueName, noAck);if (result == null) {    // No message available at this time.} else {
    IBasicProperties props = result.BasicProperties;    byte[] body = result.Body;
    ...

由于noAck = false，您還必須調(diào)用IModel.BasicAck來(lái)確認(rèn)您已成功接收并處理該消息：

  ...    // acknowledge receipt of the message
    channel.BasicAck(result.DeliveryTag, false);
}

請(qǐng)注意，使用此API獲取消息效率相對(duì)較低。如果您希望RabbitMQ將消息推送給客戶端，請(qǐng)參閱下一節(jié)。

6、通過(guò)訂閱檢索郵件（“推送API”）

接收消息的另一種方法是使用IBasicConsumer接口建立訂閱。然后，消息將在其到達(dá)時(shí)自動(dòng)發(fā)送，而不必主動(dòng)請(qǐng)求。實(shí)現(xiàn)【消費(fèi)者】的一種方法是使用便利類EventingBasicConsumer，它將傳送和其他【消費(fèi)者】生命周期事件以C＃事件：

var consumer = new EventingBasicConsumer(channel);
consumer.Received += (ch, ea) =>
                {                    var body = ea.Body;                    // ... process the message
                    channel.BasicAck(ea.DeliveryTag, false);
                };
String consumerTag = channel.BasicConsume(queueName, false, consumer);

另一個(gè)選項(xiàng)是根據(jù)子類DefaultBasicConsumer，覆蓋方法，或者直接實(shí)現(xiàn)IBasicConsumer。你一般會(huì)想實(shí)現(xiàn)核心方法IBasicConsumer.HandleBasicDeliver。更復(fù)雜的消費(fèi)者將需要實(shí)施進(jìn)一步的方法。特別地，HandleModelShutdown使通道/連接關(guān)閉。【消費(fèi)者】還可以實(shí)現(xiàn)HandleBasicCancelOk以獲得取消通知。 DefaultBasicConsumer的ConsumerTag屬性可用于檢索服務(wù)器生成的使用者標(biāo)簽，如果沒有提供給原始的IModel.BasicConsume調(diào)用。您可以使用IModel.BasicCancel取消活躍的消費(fèi)者：

channel.BasicCancel(consumerTag);

在調(diào)用API方法時(shí)，您總是通過(guò)【消費(fèi)者】標(biāo)簽來(lái)引用【消費(fèi)者】，【消費(fèi)者】標(biāo)簽可以是客戶端或服務(wù)器生成的，如AMQP 0-9-1規(guī)范文檔中所述。

7、【消費(fèi)者】的并發(fā)注意事項(xiàng)

    在當(dāng)前實(shí)現(xiàn)中，每個(gè)IConnection實(shí)例都由一個(gè)從套接字讀取的后臺(tái)線程支持，并將生成的事件分派到應(yīng)用程序。如果啟用了心跳，則從3.5.0版開始，它們將以.NET定時(shí)器方式實(shí)現(xiàn)。因此，通常使用此庫(kù)的應(yīng)用程序中至少有兩個(gè)線程處于活動(dòng)狀態(tài)：

應(yīng)用程序線程

     包含應(yīng)用程序邏輯，并調(diào)用IModel方法來(lái)執(zhí)行協(xié)議操作。

I / O活動(dòng)線程

     隱藏并由IConnection實(shí)例完全管理。

     在應(yīng)用程序中可以看到線程模型的本質(zhì)的一個(gè)地方是在庫(kù)中的應(yīng)用程序注冊(cè)的任何回調(diào)。這種回調(diào)包括：

     任何IBasicConsumer方法
     在IModel上的BasicReturn事件
     IConnection，IModel等各種關(guān)機(jī)事件

8、【消費(fèi)者】回調(diào)和訂閱

    從版本3.5.0開始，應(yīng)用程序回調(diào)處理程序可以調(diào)用阻塞操作（如IModel.QueueDeclare或IModel.BasicCancel）。 IBasicConsumer回調(diào)同時(shí)被調(diào)用。但是，每通道操作命令被保留。換句話說(shuō)，如果消息A和B按照同一個(gè)頻道的順序傳送，則按照此順序進(jìn)行處理。如果消息A和B在不同的信道上傳送，則可以以任何順序（或并行）處理消息。由【.NET運(yùn)行時(shí)】提供的、默認(rèn)的TaskScheduler的任務(wù)分發(fā)器中調(diào)用【消費(fèi)者】的回調(diào)。

9、使用自定義任務(wù)計(jì)劃程序

   通過(guò)設(shè)置ConnectionFactory.TaskScheduler可以使用自定義任務(wù)調(diào)度程序：

public class CustomTaskScheduler：TaskScheduler
{   // ...}var cf = new ConnectionFactory（）;
cf.TaskScheduler = new CustomTaskScheduler（）;

例如，這可以用于通過(guò)自定義TaskScheduler來(lái)限制并發(fā)程度。

10、在多線程中共享通道Chanel

作為經(jīng)驗(yàn)法則，IModel實(shí)例不應(yīng)同時(shí)被多個(gè)線程使用：應(yīng)用程序代碼應(yīng)該保持對(duì)IModel實(shí)例的線程所有權(quán)的清晰認(rèn)識(shí)。如果多個(gè)線程需要訪問(wèn)特定的IModel實(shí)例，應(yīng)用程序應(yīng)該強(qiáng)制執(zhí)行互斥。實(shí)現(xiàn)此目的的一種方法是為IModel的所有用戶鎖定實(shí)例本身：

IModel ch = RetrieveSomeSharedIModelInstance();lock (ch) {
  ch.BasicPublish(...);
}

IModel操作錯(cuò)誤一系列的癥狀包括但不限于，

在線上發(fā)送無(wú)效幀序列（例如，如果同時(shí)運(yùn)行多個(gè)BasicPublish操作，則發(fā)生）和/或從類RpcContinuationQueue中的方法拋出NotSupportedException異常，提示“管道的請(qǐng)求是被禁止的【Pipelining of requests forbidden】”（在同時(shí)運(yùn)行多個(gè)AMQP 0-9-1同步操作（如ExchangeDeclare）的情況下發(fā)生）。

11、處理不可路由的消息

如果發(fā)布了一個(gè)設(shè)置了“強(qiáng)制”標(biāo)志的消息，但未能送達(dá)，則代理將消息返回給發(fā)送的客戶端（通過(guò)basic.return AMQP 0-9-1命令）。要獲得此類通知，客戶端可以訂閱IModel.BasicReturn事件。如果沒有附加事件的監(jiān)聽器，則返回的消息將被靜默地丟棄。

model.BasicReturn +=  new RabbitMQ.Client.Events.BasicReturnEventHandler(...);

如果客戶端把一條標(biāo)識(shí)為“mandatory”的消息發(fā)送到了類型為【Direct】的【消息交換機(jī)Exchange】，但是這個(gè)exchange還沒有綁定到一個(gè)消息隊(duì)列的時(shí)候，BasicReturn事件就會(huì)被觸發(fā)。

12、從RabbitMQ斷開連接

要斷開連接，只需關(guān)閉通道和連接：

channel.Close(200, "Goodbye");
conn.Close();

請(qǐng)注意，關(guān)閉通道是很好的做法，但不是必要的 - 當(dāng)?shù)讓舆B接關(guān)閉時(shí)，它將自動(dòng)完成。在某些情況下，您可能希望連接上的最后一個(gè)打開的通道關(guān)閉的時(shí)候連接也關(guān)閉，要實(shí)現(xiàn)此目的，請(qǐng)將IConnection.AutoClose屬性設(shè)置為true，但僅在創(chuàng)建第一個(gè)通道后：

IConnection conn = factory.CreateConnection(...);
IModel channel = conn.CreateModel();
conn.AutoClose = true;

當(dāng)AutoClose為true時(shí)，最后關(guān)閉的通道也將導(dǎo)致連接關(guān)閉。如果在創(chuàng)建任何通道之前設(shè)置為true，則連接將在那時(shí)關(guān)閉。

好了，暫時(shí)翻譯到此吧，其實(shí)還有一些沒翻譯完，暫時(shí)就不翻譯了，有時(shí)間再繼續(xù)。

再把原文的地址貼出來(lái)，想看完整的可以通過(guò)連接瀏覽。地址如下：http://www.rabbitmq.com/dotnet-api-guide.html

天下國(guó)家,可均也；爵祿,可辭也；白刃,可蹈也；中庸不可能也

http://www.cnblogs.com/PatrickLiu/p/7193578.html

分類導(dǎo)航

RabbitMQ系列教程之七：RabbitMQ的 C# 客戶端 API 的簡(jiǎn)介

延伸閱讀

我想了解如何學(xué)習(xí)