.Net RabbitMQ 初探

前言：windows环境的安装，就不细说了，可以自行安装对应版本。查看官方对依赖版本的要求既可。

本文参考版本 RabbitMQ 3.8.7 Erlang 23.0

一、队列的常用消费模式

简单模式Simple
工作模式Work
发布订阅模式Publish/Subscribe
路由模式Routing
通配符模式Topics

二、常用模式解析

　　1.简单模式

　　　　简单的发送与接收，没有特别的处理。生产者产生数据，发送到队列，消费者拿到消息进行消费。

　　2.工作者模式

　　　　一个生产者，多个消费者。消费者一起消费队列中的数据，提高消费速度。

　　 3.发布、订阅模式

　　　　生产者端发送消息，多个消费者同时接收所有的消息。

　　4.路由模式

　　　　生产者按routing key发送消息，不同的消费者端按不同的routing key接收消息。

　　5.通配符（或主题）模式

　　　　生产者端不只按固定的routing key发送消息，而是按字符串（单词中间以点隔开）“匹配”发送，消费者端同样如此。与之前的路由模式相比，它将信息的传输类型的key更加细。

三、伪代码解析

　　 前置知识点：

　　　　AMQP模型

我们需要明确的定义服务器的语义，因为所有服务器实现都应该保持这些语义的一致性，否则就无法进行互操作。
因此AMQP模型描述了一套模块化的组件以及这些组件之间进行连接的标准规则。
在服务器中，三个主要功能模块连接成一个处理链完成预期的功能：
“exchange”接收发布应用程序发送的消息，并根据一定的规则将这些消息路由到“消息队列”。
“message queue”存储消息，直到这些消息被消费者安全处理完为止。
“binding”定义了exchange和message queue之间的关联，提供路由规则。
使用这个模型我们可以很容易的模拟出存储转发队列和主题订阅这些典型的消息中间件概念。
一个AMQP服务器类似于邮件服务器，exchange类似于消息传输代理（email里的概念），message queue类似于邮箱。Binding定义了每一个传输代理中的消息路由表，发布者将消息发给特定的传输代理，然后传输代理将这些消息路由到邮箱中，消费者从这些邮箱中取出消息。

从AMQP协议可以看出，MessageQueue、Exchange和Binding构成了AMQP协议的核心。
在rabbitmq中，exchange有4个类型：
direct，
topic，
fanout，
header。

　　1.简单模式

　　使用默认交换器，消费者操作的时候有错误，我暂时还没找到解决方案，知道了再来更新。这里用自定义交换机来变相实现，路由键默认为空，fanout 为广播类型（和发布订阅一样），会像所有绑定到该交换器上的队列发送消息

//定义交换器
channel.ExchangeDeclare("SimpelExchange", ExchangeType.Fanout, true);
//定义队列
channel.QueueDeclare("SimpelQueue", false, false, false, null);
//生成消息
channel.BasicPublish("SimpelExchange", "", properties, Encoding.UTF8.GetBytes(message));

channel.ExchangeDeclare("SimpelExchange", ExchangeType.Fanout, true);
channel.QueueDeclare("SimpelQueue", false, false, false, null);
channel.QueueBind("SimpelQueue", "SimpelExchange", "", null);
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
   var body = ea.Body;
   var message = Encoding.UTF8.GetString(body.ToArray());
   Console.WriteLine($"routing Consumer1 收到消息： {message}，时间：{DateTime.Now}");

 };
channel.BasicConsume("SimpelQueue", true, consumer);

　　2.工作者模式

　　生产者和消费者和简单模式一样，只需要限制预加载为1，同一时刻给一个消费者只有一条消息

//同一时刻服务器只发送1条消息给消费者（能者多劳，消费消息快的，会消费更多的消息）
//保证在接收端一个消息没有处理完时不会接收另一个消息，即消费者端发送了ack后才会接收下一个消息。
//在这种情况下生产者端会尝试把消息发送给下一个空闲的消费者。
channel.basicQos(1);

　　3.发布、订阅模式

　　这里用自定义交换机来变相实现，路由键默认为空，fanout 为广播类型（和发布订阅一样），会像所有绑定到该交换器上的队列发送消息

　　伪代码和简单模式一样，只需要注意当为 fanout 类型的交换器时，routing key 会被忽略

　　4.路由模式

　　交换器需要设置为 Direct 类型，生产者和消费者指定 routing key ,相同的routing key就会接受到对应的消息。

//定义交换器
channel.ExchangeDeclare("Exchange", ExchangeType.Direct, true); 
//定义队列 
channel.QueueDeclare("Queue", false, false, false, null);
//生成消息 
channel.BasicPublish("Exchange", "routingKey", properties, Encoding.UTF8.GetBytes(message));

//交换器推消息模式
channel.ExchangeDeclare("Exchange", ExchangeType.Direct, true);
channel.QueueDeclare("Queue", false, false, false, null);
channel.QueueBind("Queue", "Exchange", "routingKey", null);
var consumer = new EventingBasicConsumer(channel);
consumer.Received += (model, ea) =>
{
var body = ea.Body;
var message = Encoding.UTF8.GetString(body.ToArray());
Console.WriteLine($"routing Consumer1 收到消息： {message}，时间：{DateTime.Now}");

};
channel.BasicConsume("SimpelQueue", true, consumer);

　　注意事项

　　　　1.如果采用的交换器推消息的事件模式，需要注意，不能关闭队列通道和连接，不然事件不能持续监听。

　　　　2.ACK 确认模式，手动或是自动，处理失败的可以通过参数设置重新回到队列头（否则消息会在连接断开之前一直停留在unack，无法重新进行消费），等待重新消费，******需要注意陷入死循环*******。

//轮训拉消息模式
while (true)
{
    BasicGetResult msgResponse = channel.BasicGet("FirstQueue", false);//此处设置手动ACK确认模式
    if (msgResponse != null)
    {
        var msgBody = Encoding.UTF8.GetString(msgResponse.Body.ToArray());

        Console.WriteLine(string.Format("***接收时间:{0}，消息内容：{1}", DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss"), msgBody));
        if (msgBody.Contains("8"))
        {
　　　　　　　//手动 ACK 确认，队列会清除该条消息
            channel.BasicAck(msgResponse.DeliveryTag, false);
            Console.WriteLine("消费成功！");
        }
        else
        {
　　　　　　　//失败，消息会返回到 队列中，重新等待消费
            channel.BasicNack(msgResponse.DeliveryTag, false, true);//此处最后一个参数，可以指定是否回到队列头
        }
    }

    System.Threading.Thread.Sleep(100);
}

　　5.通配符（或主题）模式

　　交换器需要设置为 Direct 类型，生产者和消费者指定通配符,匹配的就会收到对应的消息。类似于正则匹配，规则固定为“.”分隔单词。

　　* 匹配一个单词

　　# 匹配多个单词

　　如“china.#”能够匹配到“china.news.info”，但是“china.* ”只会匹配到“china.news”

//声明交换机类型为topic
channel.exchangeDeclare(EXCHANGE_NAME, "topic");

String message = "发布了一条中国新闻消息";
channel.basicPublish(EXCHANGE_NAME, "china.news", null, message.getBytes());

message = "发布了一条中国天气消息";
channel.basicPublish(EXCHANGE_NAME, "china.weather", null, message.getBytes());

message = "发布了一条美国新闻消息";
channel.basicPublish(EXCHANGE_NAME, "usa.news", null, message.getBytes());

message = "发布了一条美国天气消息";
channel.basicPublish(EXCHANGE_NAME, "usa.weather", null, message.getBytes());

//交换机申明
channel.exchangeDeclare(EXCHANGE_NAME, "topic");
//队列声明
channel.queueDeclare(QUEUE_NAME, true, false, false, null);
//队列绑定交换机并制定路由routingKey
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "china.#");