近期，业务调整，需要内网读取数据后存入到外网，同时，其他服务器也需要读取数据,于是我又盯上了RabbitMQ。在展开业务代码前，先看下RabbitMQ整体架构，可以看到Exchange和队列是多对多关系。
下面，我们详细说说RabbitMQ的队列模式：简单队列、工作队列、发布订阅模式、路由模式、主题模式、RPC模式。其中简单队列、工作队列在前文 组件使用（一）中以提到了 ，感兴趣的可以看看，本文主要阐述 发布订阅模式、路由模式、主题模式。
- 发布订阅模式
无选择接收消息，一个消息生产者，一个交换器，多个消息队列，多个消费者。称为发布/订阅模式。在应用中，只需要简单的将队列绑定到交换机上。
一个发送到交换机的消息都会被转发到与该交换机绑定的所有队列上。像子网广播，每台子网内的主机都获得了一份复制的消息。 可以将消息发送给不同类型的消费者。做到发布一次，多个消费者来消费。
P 表示为生产者、 X 表示交换机、C1C2 表示为消费者，红色表示队列。
- 路由模式
在发布/订阅模式的基础上，有选择的接收消息，也就是通过 routing 路由进行匹配条件是否满足接收消息。
路由模式跟发布订阅模式类似，然后在订阅模式的基础上加上了类型，订阅模式是分发到所有绑定到交换机的队列，路由模式只分发到绑定在交换机上面指定路由键的队列，我们可以看一下下面这张图：
P 表示为生产者、 X 表示交换机、C1C2 表示为消费者，红色表示队列。
上图是一个结合日志消费级别的配图，在路由模式它会把消息路由到那些 binding key 与 routing key 完全匹配的 Queue 中，此模式也就是 Exchange 模式中的direct模式。 以上图的配置为例，我们以 routingKey=“error” 发送消息到 Exchange，则消息会路由到Queue1（amqp.gen-S9b…，这是由RabbitMQ自动生成的Queue名称）和Queue2（amqp.gen-Agl…）。如果我们以 routingKey=“info” 或 routingKey=“warning” 来发送消息，则消息只会路由到 Queue2。如果我们以其他 routingKey 发送消息，则消息不会路由到这两个 Queue 中。
- 主题模式
同样是在发布/订阅模式的基础上，根据主题匹配进行筛选是否接收消息，比第四类更灵活。 topics 主题模式跟 routing 路由模式类似，只不过路由模式是指定固定的路由键 routingKey，而主题模式是可以模糊匹配路由键 routingKey，类似于SQL中 = 和 like 的关系。
P 表示为生产者、 X 表示交换机、C1C2 表示为消费者，红色表示队列。
topics 模式与 routing 模式比较相近，topics 模式不能具有任意的 routingKey，必须由一个英文句点号“.”分隔的字符串（我们将被句点号“.”分隔开的每一段独立的字符串称为一个单词），比如 “lazy.orange.fox”。topics routingKey 中可以存在两种特殊字符“*”与“#”，用于做模糊匹配，其中“*”用于匹配一个单词，“#”用于匹配多个单词（可以是零个）。 "*“ 表示任何一个词 ”#" 表示0或多个词 以上图中的配置为例： 如果一个消息的 routingKey 设置为 “xxx.orange.rabbit”，那么该消息会同时路由到 Q1 与 Q2，routingKey="lazy.orange.fox”的消息会路由到Q1与Q2； routingKey="lazy.brown.fox”的消息会路由到 Q2； routingKey="lazy.pink.rabbit”的消息会路由到 Q2（只会投递给Q2一次，虽然这个routingKey 与 Q2 的两个 bindingKey 都匹配）； routingKey=“quick.brown.fox”、routingKey=”orange”、routingKey="quick.orange.male.rabbit”的消息将会被丢弃，因为它们没有匹配任何bindingKey。
- RPC模式
与上面3种所不同之处，RPC模式是拥有请求/回复的。也就是有响应的。RPC是指远程过程调用，也就是说两台服务器A，B，一个应用部署在A服务器上，想要调用B服务器上应用提供的函数/方法，由于不在一个内存空间，不能直接调用，需要通过网络来表达调用的语义和传达调用的数据。
为什么使用RPC呢？就是无法在一个进程内，甚至一个计算机内通过本地调用的方式完成的需求，比如不同的系统间的通讯，甚至不同的组织间的通讯。由于计算能力需要横向扩展，需要在多台机器组成的集群上部署应用，RPC的协议有很多，比如最早的CORBA，Java RMI，Web Service的RPC风格，Hessian，Thrift，甚至Rest API。
RPC的处理流程：
-
- 当客户端启动时，创建一个匿名的回调队列。
- 客户端为RPC请求设置2个属性：replyTo，设置回调队列名字；correlationId，标记request。
- 请求被发送到rpc_queue队列中。
- RPC服务器端监听rpc_queue队列中的请求，当请求到来时，服务器端会处理并且把带有结果的消息发送给客户端。接收的队列就是replyTo设定的回调队列。
- 客户端监听回调队列，当有消息时，检查correlationId属性，如果与request中匹配，那就是结果了。
以上就是多模式的简介，在实际生产中，我们不同模式需要定义自己交换机，其中：直接交换机、主题交换机、扇形交换机、默认交换机是常用模式。如图：
直接交换机、主题交换机、扇形交换机相关源码不多再赘述，相关代码如下：
- 连接类
public class RabbitMQConnectHelper
{
/// <summary>
/// 单方式连接
/// </summary>
/// <returns></returns>
public static IConnection GetConnection()
{
var factory = new ConnectionFactory
{
HostName = "127.0.0.1",
Port = 5672,
UserName = "gerry",
Password = "gerry",
VirtualHost = "/",//虚拟主机
};
return factory.CreateConnection();
}
/// <summary>
/// 集群方式连接
/// </summary>
/// <returns></returns>
public static IConnection GetClusterConnectiont()
{
var factory = new ConnectionFactory
{
UserName = "gerry",
Password = "gerry",
VirtualHost = "/",//虚拟主机
};
List<AmqpTcpEndpoint> host_list = new List<AmqpTcpEndpoint>
{
new AmqpTcpEndpoint(){HostName= "127.0.0.1",Port=5672},
new AmqpTcpEndpoint(){HostName= "127.0.0.1",Port=5673},
new AmqpTcpEndpoint(){HostName= "127.0.0.1",Port=5674}
};
return factory.CreateConnection(host_list);
}
}
- 生产者
public class RabbitMQ_Producer
{
/// <summary>
/// Fanout 交换机,扇形队列，数据同步发送到所有队列
/// </summary>
public void Fanout_SendMessage()
{
using (var Connection = RabbitMQConnectHelper.GetConnection())
{
using (var channel = Connection.CreateModel())
{
string queueName01 = "testQueue_01";
string queueName02 = "testQueue_02";
// 声明交换机
channel.ExchangeDeclare("test_fanout_exchange", "fanout");
//声明队列
channel.QueueDeclare(queueName01, durable: true, exclusive: false, autoDelete: false, arguments: null);
channel.QueueDeclare(queueName02, durable: true, exclusive: false, autoDelete: false, arguments: null);
//绑定到交换机
//routingKey: "" 没有绑定路由key时，消息同步到所有队列，
channel.QueueBind(queue: queueName01, exchange: "test_fanout_exchange", routingKey: "", arguments: null);
channel.QueueBind(queue: queueName02, exchange: "test_fanout_exchange", routingKey: "", arguments: null);
//声明基础属性
var properties = channel.CreateBasicProperties();
properties.AppId = Guid.NewGuid().ToString();
properties.DeliveryMode = 2;//设置持久性 1-非持久性；2-持久性
properties.Persistent = true;//设置持久化
properties.Type = queueName01;//消息类型名
properties.ContentType = "application/json";
properties.ContentEncoding = "utf-8";
for (int i = 1; i <= 20; i++)
{
string msg = $"RabbitMQ Fanout Send {i} Message!";
//routingKey: "" 没有绑定路由key时，消息同步到所有队列
channel.BasicPublish(exchange: "test_fanout_exchange", routingKey: "", basicProperties: properties, body: Encoding.UTF8.GetBytes(msg));
Console.WriteLine(msg);
}
}
}
}
/// <summary>
///Direct 交换机,直接队列，数据同步发送到特定的队列
/// </summary>
public void Direct_SendMessage()
{
using (var Connection = RabbitMQConnectHelper.GetConnection())
{
using (var channel = Connection.CreateModel())
{
string queueName01 = "testQueue_Red";
string queueName02 = "testQueue_Yellow";
// 声明交换机
channel.ExchangeDeclare("test_direct_exchange", "direct");
//声明队列
channel.QueueDeclare(queueName01, durable: true, exclusive: false, autoDelete: false, arguments: null);
channel.QueueDeclare(queueName02, durable: true, exclusive: false, autoDelete: false, arguments: null);
//绑定到交换机
//routingKey: queueName 指定消息发送到某个特定队列
channel.QueueBind(queue:queueName01, exchange: "test_direct_exchange", routingKey: queueName01, arguments: null);
channel.QueueBind(queue: queueName02, exchange: "test_direct_exchange", routingKey: queueName02, arguments: null);
//声明一个基础属性
var properties = channel.CreateBasicProperties();
properties.AppId = Guid.NewGuid().ToString();
properties.DeliveryMode = 2;//设置持久性 1-非持久性；2-持久性
properties.Persistent = true;//设置持久化
properties.Type = queueName01;//消息类型名
properties.ContentType = "application/json";
properties.ContentEncoding = "utf-8";
for (int i = 1; i <= 20; i++)
{
string msg = $"RabbitMQ Direct Send {i} Message!";
//routingKey: queueName 指定消息发送到某个特定队列 queueName01
channel.BasicPublish(exchange: "test_direct_exchange", routingKey: queueName01, basicProperties: properties, body: Encoding.UTF8.GetBytes(msg));
Console.WriteLine(msg);
}
}
}
}
/// <summary>
/// Topic 交换机,模糊队列，数据同步发送到模糊匹配队列
/// </summary>
public void Topic_SendMessage()
{
using (var Connection = RabbitMQConnectHelper.GetConnection())
{
using (var channel = Connection.CreateModel())
{
string queueName01 = "testQueue_01";
string queueName02 = "testQueue_02";
// 声明交换机
channel.ExchangeDeclare("test_topic_exchange", "topic");
//声明队列
channel.QueueDeclare(queueName01, durable: true, exclusive: false, autoDelete: false, arguments: null);
channel.QueueDeclare(queueName02, durable: true, exclusive: false, autoDelete: false, arguments: null);
//绑定到交换机
//routingKey: queueName 指定消息发送到某个特定队列
channel.QueueBind(queue:queueName01, exchange: "test_topic_exchange", routingKey: "data.*", arguments: null);
channel.QueueBind(queue: queueName02, exchange: "test_topic_exchange", routingKey: "data.Red", arguments: null);
//声明一个基础属性
var properties = channel.CreateBasicProperties();
properties.AppId = Guid.NewGuid().ToString();
properties.DeliveryMode = 2;//设置持久性 1-非持久性；2-持久性
properties.Persistent = true;//设置持久化
properties.Type = queueName01;//消息类型名
properties.ContentType = "application/json";
properties.ContentEncoding = "utf-8";
for (int i = 1; i <= 20; i++)
{
string msg = $"RabbitMQ Topic Send {i} Message!";
//routingKey: "data.Red" 指定消息发送到路由是 "data.Red"的特定队列,"data.*"属于模糊路由，也会发送数据
channel.BasicPublish(exchange: "test_topic_exchange", routingKey: "data.Red", basicProperties: properties, body: Encoding.UTF8.GetBytes(msg));
Console.WriteLine(msg);
}
}
}
}
}
- 消费者
public class RabbitMQ_Consumer
{
/// <summary>
/// fanout 消费消息数据
/// </summary>
public void fanout_Received_Message()
{
var Connection = RabbitMQConnectHelper.GetConnection();
var channel = Connection.CreateModel();
string queueName01 = "testQueue_01";
// 声明交换机
channel.ExchangeDeclare("test_fanout_exchange", "fanout");
//声明队列
channel.QueueDeclare(queueName01, durable: true, exclusive: false, autoDelete: false, arguments: null);
//绑定到交换机
//routingKey: "" 没有绑定路由key时，消息同步到所有队列，
channel.QueueBind(queue: queueName01, exchange: "test_fanout_exchange", routingKey: "", arguments: null);
//声明消费者
EventingBasicConsumer consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
var body = ea.Body;
var Str = Encoding.UTF8.GetString(body.ToArray());
//autoAck: false 则手动在接收方法内提交BasicAck 可做成等待SQL执行返回True，以保证消息能消费成功且入库
//autoAck: true 不需要在接收方法内使用 BasicAck方法
channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
};
//autoAck 值取决与 channel.BasicAck是否手动提交
channel.BasicConsume(queue: queueName01, autoAck: false, consumer: consumer);
}
/// <summary>
/// Direct 消费消息数据
/// </summary>
public void direct_Received_Message()
{
var Connection = RabbitMQConnectHelper.GetConnection();
var channel = Connection.CreateModel();
string queueName01 = "testQueue_Red";
// 声明交换机
channel.ExchangeDeclare("test_direct_exchange", "direct");
//声明队列
channel.QueueDeclare(queueName01, durable: true, exclusive: false, autoDelete: false, arguments: null);
//绑定到交换机
//routingKey: queueName 指定消息发送到某个特定队列
channel.QueueBind(queue: queueName01, exchange: "test_direct_exchange", routingKey: queueName01, arguments: null);
//声明消费者
EventingBasicConsumer consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
var body = ea.Body;
var Str = Encoding.UTF8.GetString(body.ToArray());
//autoAck: false 则手动在接收方法内提交BasicAck 可做成等待SQL执行返回True，以保证消息能消费成功且入库
//autoAck: true 不需要在接收方法内使用 BasicAck方法
channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
};
//autoAck 值取决与 channel.BasicAck是否手动提交
channel.BasicConsume(queue: queueName01, autoAck: false, consumer: consumer);
}
/// <summary>
/// Topic 消费消息数据
/// </summary>
public void topic_Received_Message()
{
var Connection = RabbitMQConnectHelper.GetConnection();
var channel = Connection.CreateModel();
string queueName02 = "testQueue_02";
// 声明交换机
channel.ExchangeDeclare("test_topic_exchange", "topic");
//声明队列
channel.QueueDeclare(queueName02, durable: true, exclusive: false, autoDelete: false, arguments: null);
//绑定到交换机
//routingKey: queueName 指定消息发送到某个特定队列
channel.QueueBind(queue: queueName02, exchange: "test_topic_exchange", routingKey: "data.Red", arguments: null);
//声明消费者
EventingBasicConsumer consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
var body = ea.Body;
var Str = Encoding.UTF8.GetString(body.ToArray());
//autoAck: false 则手动在接收方法内提交BasicAck 可做成等待SQL执行返回True，以保证消息能消费成功且入库
//autoAck: true 不需要在接收方法内使用 BasicAck方法
channel.BasicAck(deliveryTag: ea.DeliveryTag, multiple: false);
};
//autoAck 值取决与 channel.BasicAck是否手动提交
channel.BasicConsume(queue: queueName02, autoAck: false, consumer: consumer);
}
}
使用默认交换机，简洁使用。请访问基础篇 ASP.NET Core知识之RabbitMQ组件使用（一） 。