RabbitMQ --- Publish-Subscribe（发布-订阅）

目录

RabbitMQ --- Hello Mr.Tua

RabbitMQ --- Work Queues（工作队列）

前言

在第二篇文章中介绍了 Work Queues（工作队列），它适用于把一个消息发送给一个
Consumer 去处理的场景，也就是说每个
Consumer 都是处理不同的消息，如果某个 Consumer 挂了才会把未完成或没有处理的消息转发给其它的
Consumer 去处理。假如要把一个消息发送给多个 Consumer 去处理，也就是说每个
Consumer 都是处理相同的消息，这种场景就需要用到发布/订阅（Publish/Subscribe）模式，即
Producer 为发布者，Consumer 为订阅者。

匿名交换机（Nameless exchange）

Exchange 被称为交换机，担任中转站的职责，其工作原理是 Producer 先把消息发送给 Exchange，然后再由 Exchange
指定消息按照路由规则进入队列转发给 Consumer 去处理。在前两篇文章的示例中其实已经用到了默认的 Exchange，即匿名交换机（Nameless exchange）。

/*exchange:指定空字符串表示默认的Exchange标识 | routingKey:指定消息发送到哪个队列*/
channel.BasicPublish（exchange, routingKey, basicProperties, body）

交换机的类型（Exchange types）

Exchange 必须要知道如何处理它接收到的每个消息，是否让消息进入指定的队列？是否让消息进入多个队列？是否直接忽略消息？这些都取决于它的四种类型。

1.Direct：处理路由键。需要将一个队列绑定到 Exchange 上，只允许该消息与一个特定的路由键名称完整匹配才能被转发给队列。

/*Producer*/

//定义交换机
channel.ExchangeDeclare
(
exchange: "Tua",
type: ExchangeType.Direct//类型
);

channel.BasicPublish
(
exchange: "Tua",
routingKey: "Mr.Tua",//路由键名称
basicProperties: null,
body: body
);

/*Consumer*/

//定义交换机
channel.ExchangeDeclare
(
exchange: "Tua",
type: ExchangeType.Direct
);

//绑定交换机和队列，指明如何转发消息给队列
channel.QueueBind
(
queue: queueName,
exchange: "Tua",
routingKey: "Mr.Tua"//此处和生产者的路由键作比较，相同则转发消息，反之则不转发
);

2.Fanout：不处理路由键。只要将队列绑定到 Exchange 上，Producer 发送到 Exchange 的消息都会被广播到所有的队列上。 

/*Producer*/

channel.ExchangeDeclare
(
exchange: "Tua",
type: ExchangeType.Fanout
);

channel.BasicPublish
(
exchange: "Tua",
routingKey: string.Empty,
basicProperties: null,
body: body
);

/*Consumer*/

channel.ExchangeDeclare
(
exchange: "Tua",
type: ExchangeType.Fanout
);

//绑定交换机和队列，指明如何转发消息给队列
channel.QueueBind
(
queue: queueName,
exchange: "Tua",
routingKey: string.Empty
);

3.Topic：将路由键和某种命名规则进行匹配，只允许匹配上的消息才能被转发给队列。符号“#”匹配一个或多个词，符号“*”匹配一个词。

/*Producer*/

channel.ExchangeDeclare
(
exchange: "Tua",
type: ExchangeType.Topic
);

channel.BasicPublish
(
exchange: "Tua",
routingKey: "Mr.Tua.Tua"
basicProperties: null,
body: body
);

/*Consumer*/

channel.ExchangeDeclare
(
exchange: "Tua",
type: ExchangeType.Topic
);

//绑定交换机和队列，指明如何转发消息给队列
channel.QueueBind
(
queue: queueName,
exchange: "Tua",
routingKey: "Mr.#"//符号#可以匹配上Mr.Tua.Tua，符号*只能匹配上Mr.Tua
);

4.Headers：它不依赖路由键，而是在接收方绑定队列和
Exchange 时指定键值对和发送方 headers 属性中的键值对进行匹配，键值对的值可以是任何类型，只允许匹配上的消息才能被转发给队列。匹配规则
x-match 有两种类型：all 表示所有的键值对匹配；any 表示至少有一个键值对匹配。

/*Producer*/

channel.ExchangeDeclare
(
exchange: "Tua",
type: ExchangeType.Headers
);
var headers = new Dictionary<string, object>();//定义键值对集合
headers.Add("Name", "Mr.Tua");
headers.Add("Age", 102);
var basicProperties = channel.CreateBasicProperties();
basicProperties.Headers = headers;//绑定消息中的headers属性值

channel.BasicPublish
(
exchange: "Tua",
routingKey: string.Empty,
basicProperties: basicProperties,
body: body
);

/*Consumer*/

channel.ExchangeDeclare
(
exchange: "Tua",
type: ExchangeType.Headers
);
var headers = new Dictionary<string, object>();//定义键值对集合
headers.Add("x-match", "any");//定义匹配规则
headers.Add("Name", "Mr.Tua.Tua");
headers.Add("Age", 102);

channel.QueueBind
(
queue: queueName,
exchange: "Tua",
routingKey: string.Empty,
arguments: headers//指定键值对
);

 临时队列（Temporary queues）

当希望在生产者和消费者之间共享队列时，为队列命名是很重要的。如果希望接收所有最新的消息时，那么就可以不用关心队列名称了，需要一个新的空队列，而且当 Consumer 和 Rabbit 服务连接断开时需要自动删除该队列，这就是临时队列（Temporary queues）的作用。

string queueName = channel.QueueDeclare().QueueName;//创建一个随机命名的唯一非持久可以自动删除的临时队列

完整示例

现在修改上一章的代码，通过 Fanout Exchange 使两个 Consumer 接收相同的消息：

using RabbitMQ.Client;
using System;
using System.Text;

namespace Producer
{
class Program
{
static void Main(string[] args)
{
var factory = new ConnectionFactory
{
HostName = "10.202.228.107",
UserName = "Tua",
Password = "Tua",
Port = 5672
};
using (var connection = factory.CreateConnection())
{
using (var channel = connection.CreateModel())
{
channel.ExchangeDeclare
(
exchange: "Tua",
type: ExchangeType.Fanout
);
for (int m = 0; m < 10; m++)
{
string marks = string.Empty;
for (int n = 0; n <= m; n++)
{
marks += ">";
}
string msg = "Mr.Tua" + marks + marks.Length + "s";
var body = Encoding.UTF8.GetBytes(msg);
channel.BasicPublish
(
exchange: "Tua",
routingKey: "",
basicProperties: null,
body: body
);
Console.WriteLine("Producer sent message: {0}", msg);
}
Console.ReadLine();
}
}
}
}
}

Producer

using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Linq;
using System.Text;
using System.Threading;

namespace Consumer
{
class Program
{
static void Main(string[] args)
{
var factory = new ConnectionFactory
{
HostName = "localhost"
};
using (var connection = factory.CreateConnection())
{
using (var channel = connection.CreateModel())
{
channel.ExchangeDeclare
(
exchange: "Tua",
type: ExchangeType.Fanout
);
string queueName = channel.QueueDeclare().QueueName;
channel.QueueBind
(
queue: queueName,
exchange: "Tua",
routingKey: ""
);
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (sender, e) =>
{
var body = e.Body;
var msg = Encoding.UTF8.GetString(body);
int marks = msg.ToCharArray().Where(c => c.ToString() == ">").Count();
Console.WriteLine("Consumer received message: {0}", msg);
Thread.Sleep(marks * 1000);
Console.WriteLine("OK");
};
channel.BasicConsume
(
queue: queueName,
noAck: true,
consumer: consumer
);
Console.ReadLine();
}
}
}
}
}

Consumer