RabbitMQ 原文譯1.2--"Hello Word"

本系列文章均來自官網原文,屬於我的翻譯,若有雷同,權當我的歸檔,忽噴.

.NET/C# RabbitMQ 客戶端下載地址:https://github.com/rabbitmq/rabbitmq-dotnet-client

關於RabbitMQ在windows 平臺的安裝和管理配置請參考:http://www.cnblogs.com/grayguo/p/5300776.html

確保安裝成功:

這部分會寫兩個程序,一個消息生產者發送一個消息;一個消費者接受消息而後輸出到控制檯,在這個過程當中我會忽略一些.Net的細節，把注意力放在這個簡單的"hello word" 消息程序上。

在下圖中"P" 是咱們的生產者,"C" 是咱們的消費者,二者之間的中間這是咱們的消息隊列--一個隱藏在消費者後面的消息緩衝區.

發送消息:

建立一個Send.cs 來寫發送程序,發送方會鏈接RabbitMQ 服務器,發送消息,而後退出.

 

class Send
{
    public static void Main()
    {
        var factory = new ConnectionFactory() { HostName = "192.168.15.128" };
        using (var connection = factory.CreateConnection())
        {
            using (var channel = connection.CreateModel())
            {
                ...
            }
        }
    }
}

首先須要建立一個鏈接工廠去鏈接咱們的RabbitMQ服務器,這裏咱們使用RabbitMQ-dotnet-client 提供的類庫來進行回話的建立.

這裏的這裏的connection 鏈接已經爲咱們把socket 鏈接, 版本協議和認證會話,都已經爲咱們作了.這裏我鏈接的服務器是"192.168.15.128"(因爲我把RabbitMQ的測試環境搭栽了一臺虛擬機上,若是是本機可寫成"localhost"),直接指定服務器IP便可.

而後咱們在這個鏈接上建立了一次回話(channel),咱們所作的大部分Api操做都要基於會話進行。

爲了發送消息，咱們須要建立一個隊列用來存儲消息,而後能夠把咱們的消息發送到該隊列上,建立隊列代碼:

channel.QueueDeclare(queue: "hello", durable: false, exclusive: false, autoDelete: false,arguments: null);
//queue:隊名名
//durable:是否持久化
 //exclusive:是否排他
 //autoDelete:自動刪除

注:建立隊列的API是具備冪等性的--即只有當所指定的隊列不存在時纔會去建立.

而後發送消息:

 string message = "Hello World!";
 var body = Encoding.UTF8.GetBytes(message);
 channel.BasicPublish(exchange: "",routingKey: "hello", basicProperties: null, body: body);

發送的消息必須是字節數組,咱們能夠本身指定所需的編碼

完整代碼以下:

public class Send
    {
        public static void Main()
        {
            var factory = new ConnectionFactory() { HostName = "192.168.15.128" };
            using (var connection = factory.CreateConnection())
            using (var channel = connection.CreateModel())
            {
                channel.QueueDeclare(queue: "hello", durable: false, exclusive: false, autoDelete: false,
                    arguments: null);

                string message = "Hello World!";
                var body = Encoding.UTF8.GetBytes(message);

                channel.BasicPublish(exchange: "",
                                     routingKey: "hello",
                                     basicProperties: null,
                                     body: body);
                Console.WriteLine(" [x] Sent {0}", message);
            }

            Console.WriteLine(" Press [enter] to exit.");
            Console.ReadLine();
        }
    }

View Code

運行代碼,能夠經過客戶端管理工具看到結果.

 

能夠看到名字爲"hello"的隊列被建立,而且有一個消息一經存儲在隊列當中.

注:正常來講咱們的消息是需通過交換機(exchange)進行路由(route)才能到達隊列的,這裏建立完隊列而後直接(沒有手動綁定exchange和Queue)發送 routingKey爲"hello"的消息到名爲""的交換機上之因此成功,是由於當咱們建立一個隊列的時候,RabbitMQ會自動把咱們把新建的隊列和RoutingKey爲該隊列名綁定到一個默認名爲""的的交換機上。

接收消息：

RabbitMQ會主動把消息推送給咱們的消息接收者,不像消息發送者發送單個消息,咱們會讓消息接收者持續化的監聽消息而且打印出來.

 

建立一個Receive.cs 來寫接收消息的代碼

class Receive
    {
        private static void Main(string[] args)
        {
            var factory = new ConnectionFactory() { HostName = "192.168.15.128" };
            using (var connection = factory.CreateConnection())
            using (var channel = connection.CreateModel())
            {
                channel.QueueDeclare(queue: "hello",
                    durable: false,
                    exclusive: false,
                    autoDelete: false,
                    arguments: null);
            }
        }

    }

這裏的初始代碼和Send.cs 基本上是一行的,創先鏈接,建立會話,這裏之因此一樣進行名爲"hello"的隊列的建立,是爲了防止客戶端先啓動,找不到求請求的目標隊列.

鏈接服務器後,咱們告訴RabbitMQ主動把消息推送給咱們,因爲RabbitMQ推送消息是異步(asynchronously)進行的，因此咱們使用EventingBasicConsumer.Received  來進行消息的接受.

var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
                {
                    var body = ea.Body;
                    var message = Encoding.UTF8.GetString(body);
                    Console.WriteLine(" [x] Received {0}", message);
                };
channel.BasicConsume(queue: "hello",noAck: true,consumer: consumer);
//noAck(no manual acks)：ack的概念:當Consumer接收到消息、處理任務完成以後，會發送帶有這個消息標示符的ack，來告訴server這個消息接收到並處理完成.
//若是設置爲true，這個Consumer在收到消息以後會立刻返回ack(由程序自動完成 noAck=true)
//設置爲 false:須要手動發送,否者RabbitMQ會一直等處處理某個消息的Consumer的連接失去以後，才肯定這個消息沒有正確處理，從而RabbitMQ重發這個消息

完整代碼以下:

 class Receive
    {
        private static void Main(string[] args)
        {
            var factory = new ConnectionFactory() { HostName = "192.168.15.128" };
            using (var connection = factory.CreateConnection())
            using (var channel = connection.CreateModel())
            {
                channel.QueueDeclare(queue: "hello",
                    durable: false,
                    exclusive: false,
                    autoDelete: false,
                    arguments: null);

                var consumer = new EventingBasicConsumer(channel);
                consumer.Received += (model, ea) =>
                {
                    var body = ea.Body;
                    var message = Encoding.UTF8.GetString(body);
                    Console.WriteLine(" [x] Received {0}", message);
                };
                channel.BasicConsume(queue: "hello",
                    noAck: true,
                    consumer: consumer);

                Console.WriteLine(" Press [enter] to exit.");
                Console.ReadLine();
            }
        }

    }

View Code

運行代碼Send:

成功發送。

運行代碼Reveive:

消息成功接收.