RabbitMQ 原文译02

本系列文章均来自官网原文,属于个人翻译,如有雷同,权当个人归档,忽喷.

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

运行代码,可以通过客户端管理工具看到结果.

可以看到名字为"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：ack的概念:当Consumer接收到消息、处理任务完成之后，会发送带有这个消息标示符的ack，来告诉server这个消息接收到并处理完成,如果否服务器会。
//如果设置为true，这个Consumer在收到消息之后会马上返回ack
//设置为 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();      }    }  }

运行代码Send:

成功发送。

运行代码Reveive:

消息成功接收.

原标题：RabbitMQ 原文译02

关键词：