RabbitMQ入门教程——.NET客户端使用

众所周知RabbitMQ使用的是AMQP协议。我们知道AMQP是一种网络协议，能够支持符合要求的客户端应用和消息中间件代理之间进行通信。

其中消息代理扮演的角色就是从生产者那儿接受消息，并根据既定的路由规则把接受到的消息发送给消息的处理者又称消费者。由此可以看出RabbitMQ在整个消息发送，处理的过程中有三个比较重要的角色：

生产者：producer，消息生产者，就是投递消息的程序

消息代理：broker，简单来说就是消息队列服务器实体，这里简单理解为我们安装的RabbitMQ服务

消费者：consumer，消息消费者，就是接受消息的程序

接下来我们将以一个简单的控制台程序来实现消息队列的发送及接收（使用.NET版RabbitMQ客户端）：

主要功能为： 一个producer发送消息，一个consumer接收消息，并在控制台打印出来。

使用Nuget添加RabbitMQ.Client程序包至项目中

Install-Package RabbitMQ.Client

创建消息的生产者 Producer.cs ,发送一条消息给消费者

1. using RabbitMQ.Client;
2. using System;
3. using System.Text;
4.  
5. namespace RabbitMQProducer
6. {
7.     public class Producer
8.     {
9.         public static void Send()
10.         {
11.             //创建连接连接到RabbitMQ服务器，就是一个位于客户端和Broker之间的TCP连接，建议共用此TCP连接，每次使用时创建一个新的channel即可，
12.             var factory = new ConnectionFactory();
13.             IConnection connection = null;
14.             //方式1：使用AMQP协议URL amqp://username:password@hostname:port/virtual host 可通过http://127.0.0.1:15672/ RabbitMQWeb管理页面查看每个参数的具体内容
15.             factory.Uri = "amqp://guest:guest@127.0.0.1:5672//";
16.             connection = factory.CreateConnection();
17.  
18.             ////方式2：使用ConnectionFactory属性赋值
19.             //factory.UserName = ConnectionFactory.DefaultUser;
20.             //factory.Password = ConnectionFactory.DefaultPass;
21.             //factory.VirtualHost = ConnectionFactory.DefaultVHost;
22.             //factory.HostName = "127.0.0.1"; //设置RabbitMQ服务器所在的IP或主机名
23.             //factory.Port = AmqpTcpEndpoint.UseDefaultPort;
24.             //connection = factory.CreateConnection();
25.  
26.             ////方式3：使用CreateConnection方法创建连接，默认使用第一个地址连接服务端，如果第一个不可用会依次使用后面的连接
27.             //List<AmqpTcpEndpoint> endpoints = new List<AmqpTcpEndpoint>() {
28.             // new AmqpTcpEndpoint() { HostName="localhost1",Port=5672},
29.             // new AmqpTcpEndpoint() { HostName="localhost2",Port=5672},
30.             // new AmqpTcpEndpoint() { HostName="localhost3",Port=5672},
31.             // new AmqpTcpEndpoint() { HostName="localhost4",Port=5672}
32.             //};
33.             //connection = factory.CreateConnection(endpoints);
34.  
35.             using (connection)
36.             {
37.                 //创建一个消息通道，在客户端的每个连接里，可建立多个channel，每个channel代表一个会话任务。类似与Hibernate中的Session
38.                 //AMQP协议规定只有通过channel才能指定AMQP命令，所以仅仅在创建了connection后客户端还是不能发送消息的,必须要创建一个channel才行
39.                 //RabbitMQ建议客户端线程之间不要共用Channel,至少要保证共用Channel的线程发送消息必须是串行的，但是建议尽量共用Connection
40.                 using (IModel channel = connection.CreateModel())
41.                 {
42.                     //创建一个queue（消息队列）
43.                     channel.QueueDeclare(
44.                         queue: "hello",
45.                         durable: false,
46.                         exclusive: false,
47.                         autoDelete: false,
48.                         arguments: null);
49.  
50.                     string message = "你好消费者，我是生产者发送的消息";
51.  
52.                     //往队列中发出一条消息 使用了默认交换机并且绑定路由键（route key）与队列名称相同
53.                     channel.BasicPublish(
54.                         exchange: "",
55.                         routingKey: "hello",
56.                         basicProperties: null,
57.                         body: Encoding.UTF8.GetBytes(message));
58.  
59.                     Console.WriteLine($"我是生产者，我发送了一条消息{message}");
60.  
61.                     Console.WriteLine(" Press [enter] to exit.");
62.                     Console.ReadLine();
63.                 }
64.             }
65.         }
66.     }
67. }

 

注意：1.队列只会在它不存在的时候创建，多次声明并不会重复创建。

2.信息的内容是字节数组，也就意味着可以传递任何数据。

3.创建消息的消费者Consumer.cs ,从队列中获取消息并打印到屏幕

1. using RabbitMQ.Client;
2. using RabbitMQ.Client.Events;
3. using System;
4. using System.Text;
5.  
6. namespace RabbitMQConsumer
7. {
8.     public class Consumer
9.     {
10.         public static void Receive()
11.         {
12.             var factory = new ConnectionFactory();
13.             factory.Uri = "amqp://guest:guest@127.0.0.1:5672//";
14.             using (var connection = factory.CreateConnection())
15.             {
16.                 using (var channel = connection.CreateModel())
17.                 {
18.                     //声明队列，主要为了防止消息接收者先运行此程序，队列还不存在时创建队列
19.                     channel.QueueDeclare(
20.                         queue: "hello",
21.                         durable: false,
22.                         exclusive: false,
23.                         autoDelete: false,
24.                         arguments: null);
25.  
26.                     //创建事件驱动的消费者类型，尽量不要使用while(ture)循环来获取消息
27.                     EventingBasicConsumer consumer = new EventingBasicConsumer(channel);
28.                     consumer.Received += (model, ea) =>
29.                     {
30.                         var body = ea.Body;
31.                         var message = Encoding.UTF8.GetString(body);
32.                         Console.WriteLine(" 我是消费者我接收到消息： {0}", message);
33.                     };
34.  
35.                     //指定消费队列
36.                     channel.BasicConsume(queue: "hello",
37.                                          noAck: true,
38.                                          consumer: consumer);
39.  
40.                     Console.WriteLine(" Press [enter] to exit.");
41.                     Console.ReadLine();
42.                 }
43.             }
44.         }
45.     }
46. }

 

消息队列的使用过程大致如下：

• CreateConnection 创建一个连接连接到broker
• CreateModel 创建一个channel 使用它来发送AMQP指令
• ExchangeDeclare 创建一个exchange 对消息进行路由
• QueueDeclare 创建一个queue 消息队列 这是一个装载消息的容器
• QueueBind 把exchange和queue按照路由规则绑定起来
• BasicPublish 往队列中发送一条消息
• BasicConsume 从队列中获取一条消息

exchange接收到消息后，就根据消息的key和已经设置的binding，进行消息路由，将消息投递到一个或多个队列里。

本文中由于使用了默认交换机所以并没有用到 ExchangeDeclare和 QueueBind两个方法

默认交换机实际上是一个由消息代理预先声明好的没有名字（名字为空字符串）的直连交换机。它有一个特殊的属性使得它对于简单应用特别有用处：那就是每个新建队列（queue）都会自动绑定到默认交换机上，绑定的路由键（routing key）名称与队列名称相同