C#教程之C#教程之RabbitMQ基础入门篇

作者：warren

来源：http://wuchen1314.cnblogs.com/

声明：原创博客请在转载时保留原文链接或者在文章开头加上本人博客地址，如发现错误，欢迎批评指正。凡是转载于本人的文章，不能设置打赏功能，如有特殊需求请与本人联系！

下载安装

Erlang
RabbitMQ

启动RabbitMQ管理平台插件

DOS下进入到安装目录\sbin,执行以下命令

rabbitmq-plugins enable rabbitmq_management 

当出现以下结果时，重启RabbitMQ服务

set 3 plugins. Offline change; changes will take effect at broker restart.

访问http://localhost:15672（账号密码：guest）

注意：以下为C#代码，请引用NuGet包：RabbitMQ.Client

回到顶部

参考文章

RabbitMQ快速入门

回到顶部

名词解析

P（Publisher）：生产者
C（Consumer）：消费者
Channel：信道
Queue：队列
Exchange：信息交换机

回到顶部

简单演示

信息发送端

static void Send() { //1. 实例化连接工厂 var factory = new ConnectionFactory() { HostName = "localhost" }; //2. 建立连接 using (var connection = factory.CreateConnection()) { //3. 创建信道 using (var channel = connection.CreateModel()) { //4. 声明队列 channel.QueueDeclare(queue: "rabbitmq", durable: false, exclusive: false, autoDelete: false, arguments: null); //5. 构建字节数据包 var message = "Hello RabbitMQ!"; var body = Encoding.UTF8.GetBytes(message); //6. 发送数据包 channel.BasicPublish(exchange: "", routingKey: "rabbitmq", basicProperties: null, body: body); Console.WriteLine(" [x] Sent {0}", message); } } }

信息接收端

static void Receive() { //1. 实例化连接工厂 var factory = new ConnectionFactory() { HostName = "localhost" }; //2. 建立连接 using (var connection = factory.CreateConnection()) { //3. 创建信道 using (var channel = connection.CreateModel()) { //4. 声明队列 channel.QueueDeclare(queue: "rabbitmq", durable: false, exclusive: false, autoDelete: false, arguments: null); //5. 构造消费者实例 var consumer = new EventingBasicConsumer(channel); //6. 绑定消息接收后的事件委托 consumer.Received += (model, ea) => { var message = Encoding.UTF8.GetString(ea.Body); Console.WriteLine(" [x] Received {0}", message); }; //7. 启动消费者 channel.BasicConsume(queue: "rabbitmq", autoAck: true, consumer: consumer); Console.WriteLine(" Press [enter] to exit."); Console.ReadLine(); } } }

回到顶部

轮询调度

P生产的多个任务进入到队列中，多个C间可以并行处理任务。默认情况下，RabbitMQ把信息按顺序发送给每一个C。平均每个C将获得同等数量的信息。

回到顶部

信息确认

按照最简单的演示来说，信息一旦发送到C中，则该信息就会从队列中移除。一旦中间信息处理异常/失败，C端程序退出等，都将会导致信息未处理完成，而此时队列中已将信息移除了，那么就会导致一系列的问题。我们可以在C端设置手动确认信息，从而解决上述问题的发生。
Receive代码块

//6. 绑定消息接收后的事件委托 consumer.Received += (model, ea) => { var message = Encoding.UTF8.GetString(ea.Body); Console.WriteLine(" [x] Received {0}", message); Thread.Sleep(5000);//模拟耗时 Console.WriteLine(" [x] Done"); // 发送信息确认信号（手动信息确认） channel.BasicAck(ea.DeliveryTag, false); }; //7. 启动消费者 /* autoAck参数属性 true：自动信息确认，当C端接收到信息后，自动发送ack信号，不管信息是否处理完毕 false：关闭自动信息确认，通过调用BasicAck方法手动进行信息确认 */ channel.BasicConsume(queue: "rabbitmq", autoAck: false, consumer: consumer); 

回到顶部

信息持久化

当RabbitMQ退出或死机时会清空队列和信息。通过将队列和信息标记为持久的，来告知RabbitMQ将信息持久化。

Send代码块

//4. 声明队列 //durable设置为true，表示此队列为持久的。 //注意：RabbitMQ不允许你使用不同的参数重新定义一个已经存在的队列，所以你需要重启服务/更改队列名称 channel.QueueDeclare(queue: "rabbitmq", durable: true, //标记队列持久 exclusive: false, autoDelete: false, arguments: null); //设置IbasicProperties.SetPersistent属性值为true来标记我们的消息持久化 var properties = channel.CreateBasicProperties(); properties.Persistent = true; //5. 构建字节数据包 var message = "Hello RabbitMQ!"; var body = Encoding.UTF8.GetBytes(message); //6. 发送数据包 channel.BasicPublish(exchange: "", routingKey: "rabbitmq", basicProperties: properties, //指定BasicProperties body: body);

回到顶部

公平调度

上述演示中，如果队列中存在多个信息，在开启多个C的情况下，只有一个C忙个不停，另外的却一直处于空闲状态。通过调用BasicQos，告知RabbitMQ在某个C信息处理完毕，并且已经收到信息确认之后，才可以继续发送信息到这个C。否则，将会把信息分发到另外空闲的C。

Receive代码块

//4. 声明队列 channel.QueueDeclare(queue: "rabbitmq",  durable: true,  exclusive: false,  autoDelete: false,  arguments: null); ////设置prefetchCount为1来告知RabbitMQ在未收到消费端的消息确认时，不再分发消息 channel.BasicQos(prefetchSize: 0,  prefetchCount: 1,  global: false);

回到顶部

发布/订阅

上述中的演示，P推送信息至队列中，C从队列中处理信息。但是如果需要将P推送的信息至每个订阅的C中处理信息，那么我们就可以使用Exchange。

fanout（将信息分发到exchange上绑定的所有队列上）

Send代码块

//1. 实例化连接工厂 var factory = new ConnectionFactory() { HostName = "localhost" }; //2. 建立连接 using (var connection = factory.CreateConnection()) { //3. 创建信道 using (var channel = connection.CreateModel()) { //4. 声明信息交换机 channel.ExchangeDeclare(exchange: "fanoutDemo", type: "fanout"); //5. 构建字节数据包 var message = "Hello RabbitMQ!"; var body = Encoding.UTF8.GetBytes(message); //6. 发布到指定exchange，fanout类型的会忽视routingKey的值，所以无需填写 channel.BasicPublish(exchange: "fanoutDemo", routingKey: "", basicProperties: null, body: body); Console.WriteLine(" [x] Sent {0}", message); } }

Receive代码块

//1. 实例化连接工厂 var factory = new ConnectionFactory() { HostName = "localhost" }; //2. 建立连接 using (var connection = factory.CreateConnection()) { //3. 创建信道 using (var channel = connection.CreateModel()) { //4. 声明信息交换机 channel.ExchangeDeclare(exchange: "fanoutDemo", type: "fanout"); //生成随机队列名称 var queueName = channel.QueueDeclare().QueueName; //绑定队列到指定fanout类型exchange channel.QueueBind(queue: queueName, exchange: "fanoutDemo", routingKey: ""); //5. 构造消费者实例 var consumer = new EventingBasicConsumer(channel); //6. 绑定消息接收后的事件委托 consumer.Received += (model, ea) => { var message = Encoding.UTF8.GetString(ea.Body); Console.WriteLine(" [x] Received {0}", message); }; channel.BasicConsume(queue: queueName, autoAck: true, consumer: consumer); Console.WriteLine(" Press [enter] to exit."); Console.ReadLine(); } }

direct（C绑定的队列名称须和P发布指定的路由名称一致）

Send代码块

//4. 声明信息交换机 channel.ExchangeDeclare(exchange: "directDemo", type: "direct"); //5. 构建字节数据包 var message = "Hello RabbitMQ!"; var body = Encoding.UTF8.GetBytes(message); //6. 发布到指定exchange channel.BasicPublish(exchange: "directDemo", routingKey: "a", basicProperties: null, body: body);

Receive代码块

//4. 声明信息交换机 channel.ExchangeDeclare(exchange: "directDemo", type: "direct"); //生成随机队列名称 var queueName = channel.QueueDeclare().QueueName; //绑定队列到指定direct类型exchange channel.QueueBind(queue: queueName, exchange: "directDemo", routingKey: "b");

topic（支持通配符的路由规则）

通配字符：

• *：匹配一个单词
• #：匹配0个或多个单词
• .：仅作为分隔符

Send代码块

//4. 声明信息交换机 channel.ExchangeDeclare(exchange: "topicDemo", type: "topic"); //5. 构建字节数据包 var message = "Hello RabbitMQ!"; var body = Encoding.UTF8.GetBytes(message); //6. 发布到指定exchange channel.BasicPublish(exchange: "topicDemo", routingKey: "admin.user.error", //模拟后台用户错误 basicProperties: null, body: body);

Receive代码块

//4. 声明信息交换机 channel.ExchangeDeclare(exchange: "topicDemo", type: "topic"); //生成随机队列名称 var queueName = channel.QueueDeclare().QueueName; //绑定队列到指定topic类型exchange channel.QueueBind(queue: queueName, exchange: "topicDemo", routingKey: "admin.*.#"); //订阅所有后台异常错误

回到顶部

RPC（远程过程调用）

1. 进行远程调用的客户端需要指定接收远程回调的队列，并申明消费者监听此队列。
2. 远程调用的服务端除了要申明消费端接收远程调用请求外，还要将结果发送到客户端用来监听的结果的队列中去。

客户端代码块

var factory = new ConnectionFactory() { HostName = "localhost" }; using (var connection = factory.CreateConnection()) { using (var channel = connection.CreateModel()) { var correlationId = Guid.NewGuid().ToString(); var replyQueue = channel.QueueDeclare().QueueName; var properties = channel.CreateBasicProperties(); properties.ReplyTo = replyQueue; properties.CorrelationId = correlationId; string number = args.Length > 0 ? args[0] : "30"; var body = Encoding.UTF8.GetBytes(number); //发布消息 channel.BasicPublish(exchange: "", routingKey: "rpc_queue", basicProperties: properties, body: body); Console.WriteLine($"[*] Request fib({number})"); // //创建消费者用于消息回调 var callbackConsumer = new EventingBasicConsumer(channel); channel.BasicConsume(queue: replyQueue, autoAck: true, consumer: callbackConsumer); callbackConsumer.Received += (model, ea) => { if (ea.BasicProperties.CorrelationId == correlationId) { var responseMsg = $"Get Response: {Encoding.UTF8.GetString(ea.Body)}"; Console.WriteLine($"[x]: {responseMsg}"); } }; Console.ReadLine(); } }

服务端代码块

static void Main(string[] args) { var factory = new ConnectionFactory() { HostName = "localhost" }; using (var conection = factory.CreateConnection()) { using (var channel = conection.CreateModel()) { channel.QueueDeclare(queue: "rpc_queue", durable: false, exclusive: false, autoDelete: false, arguments: null); var consumer = new EventingBasicConsumer(channel); Console.WriteLine("[*] Waiting for message."); consumer.Received += (model, ea) => { var message = Encoding.UTF8.GetString(ea.Body); int n = int.Parse(message); Console.WriteLine($"Receive request of Fib({n})"); int result = Fib(n); var properties = ea.BasicProperties; var replyProerties = channel.CreateBasicProperties(); replyProerties.CorrelationId = properties.CorrelationId; channel.BasicPublish(exchange: "", routingKey: properties.ReplyTo, basicProperties: replyProerties, body: Encoding.UTF8.GetBytes(result.ToString())); channel.BasicAck(ea.DeliveryTag, false); Console.WriteLine($"Return result: Fib({n})= {result}"); }; channel.BasicConsume(queue: "rpc_queue", autoAck: false, consumer: consumer); Console.ReadLine(); } } } private static int Fib(int n) { if (n == 0 || n == 1) { return n; } return Fib(n - 1) + Fib(n - 2); }