分布式系统消息中间件——RabbitMQ的使用进阶篇 (3)

    RabbitMQ本身并未提供延迟队列的功能。延迟队列是一个逻辑上的概念，可以通过过期时间+死信队列来模拟它的实现。延迟队列的逻辑架构大致如下:

    生产者将消息发送到过期时间为n的队列中，这个队列并未有消费者来消费消息，当过期时间到达时，消息会通过死信交换器被转发到死信队列中。而消费者从死信队列中消费消息。这个时候就达到了生产者发布了消息在讲过了n时间后消费者消费了消息，起到了延迟消费的作用。

     延迟队列在我们的项目中可以应用于很多场景，如：下单后两个消息取消订单，七天自动收货，七天自动好评，密码冻结后24小时解冻，以及在分布式系统中消息补偿机制(1s后补偿,10s后补偿，5m后补偿......)。

六 优先级队列

    就像我们生活中的“特殊”人士一样，我们的业务上也存在一些“特殊”消息，可能需要优先进行处理，在生活上我们可能会对这部分特殊人士开辟一套VIP通道，而Rabbit同样也有这样的VIP通道(前提是在3.5的版本以后)，即优先级队列，队列中的消息会有优先级优先级高的消息具备优先被消费的特权。针对这些VIP消息，我们只需做两件事:

我们只需做两件事情：

将队列声明为优先级队列，即在创建队列的时候添加参数 x-max-priority 以指定最大的优先级，值为0-255（整数）。

为优先级消息添加优先级。

其示例代码如下:

channel.ExchangeDeclare("exchange.priority", "direct", true);//定义交换器 IDictionary<String, Object> args = new Dictionary<String, Object>(); args.Add("x-max-priority", 10);//定义优先级队列的最大优先级为10 channel.QueueDeclare("queue.priority", true, false, false, args);//定义优先级队列 channel.QueueBind("queue.priority", "exchange.priority", "priorityKey");//队列交换器绑定 BasicProperties properties = new BasicProperties() { Priority =8,//设置消息优先级为8 }; var message = Encoding.UTF8.GetBytes("TestMsg8"); //发布消息 channel.BasicPublish("exchange.priority", "priorityKey", properties, message);

注意：没有指定优先级的消息会将优先级以0对待。 对于超过优先级队列所定最大优先级的消息，优先级以最大优先级对待。对于相同优先级的消息，后进的排在前面。如果在消费者的消费速度大于生产者的速度且Broker 中没有消息堆积的情况下， 对发送的消息设置优先级也就没有什么实际意义。因为生产者刚发送完一条消息就被消费者消费了，那么就相当于Broker 中至多只有一条消息，对于单条消息来说优先级是没有什么意义的。

    关于优先级队列，好像违背了队列这种数据结构先进先出的原则，其具体是怎么实现的在这里就不过多讨论。有兴趣的可以自己研究研究。后续可能也会有相关的文章来分析其原理。

七 RPC 实现

    RPC,是Remote Procedure Call 的简称，即远程过程调用。它是一种通过网络从远程计算机上请求服务，而不需要了解底层网络的技术。RPC 的主要功用是让构建分布式计算更容易，在提供强大的远程调用能力时不损失本地调用的语义简洁性。

    有关RPC不多介绍，这里我们主要介绍RabbitMQ如何实现RPC。RabbitMQ 可以实现很简单的RPC。客户端发送请求消息，服务端回复响应的消息，为了接收响应的消息，我们需要在请求消息中发送一个回调队列(可以使用默认的队列)。其服务器端实现代码如下:

static void Main(string[] args) { ConnectionFactory factory = new ConnectionFactory(); factory.UserName = "admin"; factory.Password = "admin"; factory.HostName = "192.168.121.205"; IConnection conn = factory.CreateConnection(); IModel channel = conn.CreateModel(); channel.QueueDeclare("RpcQueue", true, false, false, null); SimpleRpcServer rpc = new MySimpRpcServer(new Subscription(channel, "RpcQueue")); rpc.MainLoop(); } public class MySimpRpcServer: SimpleRpcServer { public MySimpRpcServer(Subscription subscription) : base(subscription) { } /// <summary> /// 执行完成后进行回调 /// </summary> public override byte[] HandleSimpleCall(bool isRedelivered, IBasicProperties requestProperties, byte[] body, out IBasicProperties replyProperties) { replyProperties = null; return Encoding.UTF8.GetBytes("我收到了!"); } /// <summary> /// 进行处理 /// </summary> /// <param></param> public override void ProcessRequest(BasicDeliverEventArgs evt) { // todo..... base.ProcessRequest(evt); } }