RabbitMQ死信队列另类用法之复合死信

梁规晓

发布于 2019-04-22 17:23:06

1.2K0

发布于 2019-04-22 17:23:06

文章被收录于专栏：DotNet程序园

前言

在业务开发过程中，我们常常需要做一些定时任务，这些任务一般用来做监控或者清理任务，比如在订单的业务场景中，用户在创建订单后一段时间内，没有完成支付，系统将自动取消该订单，并将库存返回到商品中，又比如在微信中，用户发出红包24小时后，需要对红包进行检查，是否已领取完成，如未领取完成，将剩余金额退回到发送者钱包中，同时销毁该红包。

在项目初始阶段，或者是一些小型的项目中，常常采用定时轮询的方法进行检查，但是我们都知道，定时轮询将给数据库带来不小的压力，而且定时间隔无法进行动态调整，特别是一个系统中，同时存在好几个定时器的时候，就显得非常的麻烦，同时给数据库造成巨大的访问压力。

下面，本文将演示如何使用一个 RabbitMQ 的死信队列同时监控多种业务（复合业务），达到模块解耦，释放压力的目的。

注意：名词“复合死信”是为了叙述方便临时创造的，如有不妥，欢迎指正

1. 什么是 RabbitMQ 死信队列

DLX（Dead Letter Exchanges）死信交换，死信队列本身也是一个普通的消息队列，在创建队列的时候，通过设置一些关键参数，可以将一个普通的消息队列设置为死信队列，与其它消息队列不同的是，其入栈的消息根据入栈时指定的过期时间/被拒绝/超出队列长度被移除，依次被转发到指定的消息队列中进行二次处理。这样说法比较拗口，其原理就是死信队列内位于顶部的消息过期时，该消息将被马上发送到另外一个订阅者（消息队列）中。

其原理入下图

由上图可以看到，目前有三种类型的业务需要使用 DLX 进行处理，因为每个业务的超时时间不一致的问题，如果将他们都放入一个 DLX 中进行处理，将会出现一个时序的问题，即消息队列总数处理顶部的消息，如果顶部的消息未过期，而底部的消息过期，这就麻烦了，因为过期的消息无法得到消费，将会造成延迟；所以正常情况下，最好的办法是每个业务都独立一个队列，这样就可以保证，即将过期的消息总是处于队列的顶部，从而被第一时间处理。

但是多个 DLX 又带来了管理上面的问题，随着业务的增加，越来越多的业务需要进入不同的 DLX ，这个时候我们发现，由于人手不足的原因，维护这么多 DLX 实在是太吃力了，如果能将这些消息都接入一个 DLX 中该多好呀，在一个 DLX 中进行消息订阅，然后进行分发或者处理，这就非常有趣了。

下面就按照这个思路，我们进行集中处理,也就是复合死信交换 CDLX（Composite Dead Letter Exchanges)

2. 如何创建死信队列

创建 DLX 队列的方式非常简单，我们使用 RabbitMQ Web 控制面板进行创建 Exhcange（交换机）/Consumer（死信消费队列）/cdlx（复合死信队列）

2.1 创建队列

创建交换机 cdlx-Exchange

死信消费队列 cdlx-Consumer

复合死信队列 cdlx-Master

注意，这里添加死信队列必须同时设置死信转发交换机和路由，后续通过路由绑定实现消费队列

路由绑定

上面的路由绑定共有两个，分别是 Master 和 Consumer 用于消息路由到队列，为下面的业务消息做准备，建好后的队列如下

3.复合业务进入死信队列

当建立好队列以后，我们就可以专心的处理业务了，下面就来模拟3种业务将消息发送到死信队列的过程

3.1 发送死信消息到队列

发送消息使用了 Asp.NetCore轻松学-实现一个轻量级高可复用的RabbitMQ客户端中的轻量客户端，封装后的发送消息代码如下

public class CdlxMasterService    {        private IConfiguration cfg = null;        private ILogger logger = null;        private string vhost = "test_mq";        private string exchange = "cdlx-Exchange";        private string routekey = "master";        private static MQConnection connection = null;
        private MQConnection Connection        {            get            {                if (connection == null || !connection.Connection.IsOpen)                {                    connection = new MQConnection(                        cfg["rabbitmq:username"],                        cfg["rabbitmq:password"],                        cfg["rabbitmq:host"],                        Convert.ToInt32(cfg["rabbitmq:port"]),                        vhost,                        logger);                }                return connection;            }        }
        private static IModel channel = null;        private IModel Channel        {            get            {                if (channel == null || channel.IsClosed)                    channel = Connection.Connection.CreateModel();
                return channel;            }        }
        public void SendMessage(object data)        {            string message = JsonConvert.SerializeObject(data);            this.Connection.Publish(this.Channel, exchange, routekey, message);        }    }

3.2 将 CdlxMasterService 注入到服务

  public void ConfigureServices(IServiceCollection services)        {           services.AddSingleton<CdlxMasterService>();           ...        }

3.3 模拟3种业务生产死信消息

    public class HomeController : Controller    {        private CdlxMasterService masterService;        public HomeController(CdlxMasterService masterService)        {            this.masterService = masterService;        }
        [HttpGet("publish")]        public int Publish()        {            Contract contract = new Contract(this.masterService);            for (int i = 0; i < 10; i++)            {                contract.Publish(MessageType.RedPackage, "红包信息，超时时间1024s");                contract.Publish(MessageType.Order, "订单信息，超时时间2048s");                contract.Publish(MessageType.Vote, "投票信息，超时时间4096s");            }            return 0;        }    }

上面的接口 puhlish 模拟了业务消息，由于我们依次发布了红包/订单/投票消息，所以迭代发布 10 次后，正好形成了一个时序错乱的信息队列，按照自动过期时序计算，当第一个红包超时到达时，第四条消息（红包）也会接着超时，可是由于此时订单和投票消息位于红包消息上面，该红包消息在达到超时时间后并不会被投递到 Consumer 消费队列，这是正确的，我们确实也是希望是这个结果

如果有一个办法把超时的消息自动将其提升到队列顶部就好了！

4. 处理复合死信

在 RabbitMQ 提供的 API 接口中，没有什么直接可用的能将死信队列中超时消息提升到顶部的好办法；但是，我们可以利用部分 API 接口的特性来完成这件事情。

4.1 定时消费客户端

下面，我们将使用一个定时消费客户端来完成对死信队列的轮询，充分利用 RabbitMQ 的消费特性来完成超时消息的位置提升。

过程如下图：

如上图所示，我们增加一个 dlx-timer 定时器，定时的发起对死信队列的消费，该消费者仅仅是消费，不确认消息，也就是不做 ack，然后将消息重新置入队列中；这个过程，就是将消息不断提升位置的过程。

4.2 定时消费客户端实现代码

    public class CdlxTimerService : MQServiceBase    {        public override string vHost { get { return "test_mq"; } }        public override string Exchange { get { return "cdlx-Exchange"; } }        public override List<BindInfo> Binds => new List<BindInfo>();        private string queue = "cdlx-Master";
        public CdlxTimerService(IConfiguration cfg, ILogger logger) : base(cfg, logger)        {        }
        /// <summary>        ///  检查死信队列        /// </summary>        /// <returns></returns>        public List<CdlxMessage> CheckMessage()        {            long total = 0;            List<CdlxMessage> list = new List<CdlxMessage>();            var connection = base.CreateConnection();            using (IModel channel = connection.Connection.CreateModel())            {                bool latest = true;                while (latest)                {                    BasicGetResult result = channel.BasicGet(this.queue, false);                    total++;                    latest = result != null;                    if (latest)                    {                        var json = Encoding.UTF8.GetString(result.Body);                        list.Add(JsonConvert.DeserializeObject<CdlxMessage>(json));                    }                }                channel.Close();                connection.Close();            }            return list;        }    }

上面的代码首先在定时调用到来的时候，创建了一个 Connection，然后利用此 Connection 创建了了一个 Channel，紧接着，使用该 Channel 调用 BasicGet 方法，获得队列顶部的信息，且设置 autoAck=false，表示仅检查消息，不确认，然后进入一个 while 迭代过程，一直读取到队列底部，获得所有队列中的信息，最后，关闭了通道释放连接。

这样，就完成了一次消息检查的过程，在调用 BasicGet 后，下一条信息将会出现在队列的顶部，同步，队列将自动对该消息进行超时检查，由于我们在调用 BasicGet 的时候，传入 autoAck=false，不确认该消息，在 RabbitMQ 控制台中，将显示为 unacted，所以在释放连接后，所有消息将会被重新置入队列中，这是一个自动的过程，无需我们做额外的工作。

4.3 Consumer（死信消费队列）最终处理业务

配置队列管理随程序启动停止

        private MQServcieManager serviceManager;        // This method gets called by the runtime. Use this method to configure the HTTP request pipeline.        public void Configure(IApplicationBuilder app, IHostingEnvironment env, ILoggerFactory factory, IApplicationLifetime lifeTime)        {            serviceManager = new MQServcieManager(this.Configuration, factory.CreateLogger<MQServcieManager>());            lifeTime.ApplicationStarted.Register(() => { serviceManager.Start(); });            lifeTime.ApplicationStopping.Register(() => { serviceManager.Stop(); });            ...        }

实现消费队列

    public class CdlxConsumerService : MQServiceBase    {        public override string vHost { get { return "test_mq"; } }        public override string Exchange { get { return "cdlx-Exchange"; } }        private string queue = "cdlx-Consumer";        private string routeKey = "all";        private List<BindInfo> bs = new List<BindInfo>();        public override List<BindInfo> Binds { get { return bs; } }
        public CdlxConsumerService(IConfiguration cfg, ILogger logger) : base(cfg, logger)        {            this.bs.Add(new BindInfo            {                ExchangeType = ExchangeType.Direct,                Queue = this.queue,                RouterKey = this.routeKey,                OnReceived = this.OnReceived            });        }
        private void OnReceived(MessageBody body)        {            var message = JsonConvert.DeserializeObject<CdlxMessage>(body.Content);            Console.WriteLine("类型：{0}\t 内容：{1}\t进入时间：{2}\t过期时间：{3}", message.Type, message.Data, message.CreateTime, message.CreateTime.AddSeconds(message.Expire));
            body.Consumer.Model.BasicAck(body.BasicDeliver.DeliveryTag, true);        }    }

上面的代码，模拟了最终业务处理的过程，这里仅仅是简单演示，所以只是将消息打印到屏幕上；在实际的业务场景中，我们可以根据不同的 MessageType 进行消息的分发处理。

5. 消费过程演示

为了比较直观的观看死信消费过程，我们编写一个简单的列表页面，自动刷新后去消费死信队列，然后将消息输出到页面上，通过观察此页面，我们可以实时了解到死信队列的消费过程，实际的业务场景中，大家可以利用第三方定时器定时调用接口实现，或者使用内置的轻量主机做后台任务实现定时轮询，具体参考 Asp.Net Core 轻松学-基于微服务的后台任务调度管理器

5.1 发布消息

浏览器访问本机地址：http://localhost:5000/home/publish

下面将发布 30 条信息到 DLX 中，每个业务各 10 条信息。

通常情况下，红包的过期时间最短且超时时间一致，应该最快超时，意味着当第一条红包消息超时的时候，其余 9 条红包消息也会一并超时，但是由于红包消息混合的发布在队列中，且只有第一条红包消息位移队列顶部；所以，当第一条红包消息超时被消费后，其余 9 条红包由于不是位于队列顶部，虽然此时他们已经超时，但是 DLX 将无法处理；当我们使用 cdlx-timer（定时器）模拟调用 CdlxTimerService 的时候（也就是刷新首页）， CdlxTimerService 服务将会对 DLX 进行检查。

查看消费状态