最近遇到了这样的场景：每隔一段时间，需要在后台使用队列对一批数据进行业务处理。

Quartz.NET是一种选择，在 .NET Core中，可以使用IHostedService执行后台定时任务。在本篇中，首先尝试把队列还原到最简单、原始的状态，然后给出以上场景问题的具体解决方案。

假设一个队列有8个元素。现在abcd依次进入队列。

ab依次出队列。

可以想象，随着不断地入列出列，head和tail的位置不断往后，当tail在7号位的时候，虽然队列里还有空间，但此时数据就无法入队列了。

如何才可以继续入队列呢？

首先想到的是数据搬移。当数据无法进入队列，首先让队列项出列，进入到另外一个新队列，这个新队列就可以再次接收数据入队列了。但是，搬移整个队列中的数据的时间复杂度为O(n)，而原先出队列的时间复杂度是O(1)，这种方式不够理想。

还有一种思路是使用循环队列。当tail指向最后一个位置，此时有新的数据进入队列，tail就来到头部指向0号位置，这样这个队列就可以循环使用了。

现在a入栈。

队列有很多种实现。

比如在生产消费模型中可以用阻塞队列。当生产队列为空的时候，为了不让消费者取数据，生产队列的Dequeue行为会被阻塞；而当生产队列满的时候，为了不让更多的数据进来，生产队列的Enqueue行为被阻塞。

线程安全的队列叫并发队列，如C#中的ConcurrentQueue。

线程池内部也使用了队列机制。因为CPU的资源是有限的，过多的线程会导致CPU频繁地在线程之间切换。线程池内通过维护一定数量的线程来减轻CPU的负担。当线程池没有多余的线程可供使用，请求过来，一种方式是拒绝请求，另外一种方式是让请求队列阻塞，等到线程池内有线程可供使用，请求队列出列执行。用链表实现的队列是无界队列(unbounded queue)，这种做法可能会导致过多的请求在排队，等待响应时间过长。用数组实现的队列是有界队列(bounded queue)，当线程池已满，请求过来就会被拒绝。对有界队列来说数组的大小设置很讲究。

来模拟一个数组队列。

public class ArrayQueue{ private string[] items; private int n = 0; //数组长度 private int head = 0; private int tail = 0; public ArrayQueue(int capacity) { n = capacity; items = new string[capacity]; } public bool Enqueue(string item) { if(tail==n){ return false; } items[tail] = item; ++tail; return true; } public string Dequeue() { if(head==null){ return null; } string ret = items[head]; ++head; return ret; }}

以上就是一个最简单的、用数组实现的队列。

再次回到要解决的场景问题。解决思路大致是：实现IHostedService接口，在其中执行定时任务，每次把队列项放到队列中，并定义出队列的方法，在其中执行业务逻辑。

关于队列，通过以下的步骤使其在后台运行。

• 队列项(MessageQueueItem):具备唯一标识、委托、添加到队列中的时间等属性
• 队列(MessageQueue):维护着Dictionary<string, MessageQueueItem>静态字典集合
• MessageQueueUtility类:决定着如何运行，比如队列执行的间隔时间、垃圾回收
• MessageQueueThreadUtility类:维护队列线程，提供队列在后台运行的方法
• 在Startup.cs中的Configure中调用MessageQueueThreadUtility中的方法使队列在后台运行

队列项(MessageQueueItem)

public class MessageQueueItem{ public MessageQueueItem(string key, Action action, string description=null) { Key = key; Action = action; Description = description; AddTime = DateTime.Now; } public string Key{get;set;} public Actioin Action{get;set;} public DateTime AddTime{get;set;} public string Description{get;set;}}

队列(MessageQueue),维护着针对队列项的一个静态字典集合。

public class MessageQueue{ public static Dictionary<string, MessageQueueItem> MessageQueueDictionary = new Dictionary<string, MessageQueueItem>(StringComparer.OrdinalIgnoreCase); public static object MessageQueueSyncLock = new object(); public static object OperateLock = new object(); public static void OperateQueue() { lock(OperateLock) { var mq = new MessageQueue(); var key = mq.GetCurrentKey(); while(!string.IsNullOrEmpty(key)) { var mqItem = mq.GetItem(key); mqItem.Action(); mq.Remove(key); key = mq.GetCurrentKey(); } } } public string GetCurrentKey() { lock(MessageQueueSyncLock) { return MessageQueueDictionary.Keys.FirstOrDefault(); } } public MessageQueueItem GetItem(string key) { lock(MessageQueueSyncLock) { if(MessageQueueDictionary.ContainsKey(key)) { return MessageQueueDictionary[key]; } return null; } } public void Remove(string key) { lock(MessageQueueSyncLock) { if(MessageQueueDictionary.ContainsKey(key)) { MessageQueueDictionary.Remove(key); } } } public MessageQueueItem Add(string key, Action actioin) { lock(MessageQueueSyncLock) { var mqItem = new MessageQueueItem(key, action); MessageQueueDictionary[key] = mqItem; return mqItem; } } public int GetCount() { lock(MessageQueueSyncLock) { return MessageQueueDictionary.Count; } }}

MessageQueueUtility类, 决定着队列运行的节奏。

public class MessageQueueUtility{ private readonly int _sleepMilliSeconds; public MessageQueueUtility(int sleepMilliSeconds=1000) { _sleepMilliSeconds = sleepMilliSeoncds; } ~MessageQueueUtility() { MessageQueue.OperateQueue(); } public void Run { do { MessageQueue.OperateQueue(); Thread.Sleep(_sleepMilliSeconds); } while(true) }}

MessageQueueThreadUtility类,管理队列的线程，并让其在后台运行。

public static class MessageQueueThreadUtility{ public static Dictionary<string, Thread> AsyncThreadCollection = new Dictioanry<string, Thread>(); public static void Register(string threadUniqueName) { { MessageQueueUtility messageQueueUtility = new MessageQueueUtility(); Thread messageQueueThread = new Thread(messageQueueUtility.Run){ Name = threadUniqueName }; AsyncThreadCollection.Add(messageQueueThread.Name, messageQueueThread); } AsyncThreadCollection.Values.ToList().ForEach(z => { z.IsBackground = true; z.Start(); }); }}

Startup.cs中注册。

public class Startup{ public IServiceProvider ConfigureServices(IServiceCollection services) { ... } public void Configure(IApplicationBuilder app, IHostingEnvironment env...) { RegisterMessageQueueThreads(); } private void RegisterMessageQueueThreads() { MessageQueueThreadUtility.Register(""); }}

最后在IHostedService的实现类中把队列项丢给队列。

public class MyBackgroundSerivce : IHostedService, IDisposable{ private Timer _timer; public IServiceProvider Services{get;} public MyBackgroundService(IServiceProvider services) { Serivces = services; } public void Dispose() { _timer?.Dispose(); } public Task StartAsync(CancellationToken cancellationToken) { _timer = new Timer(DoWork, null, TimeSpan.Zero, TimeSpan.FromSeconds(10)); return Task.CompletedTask; } public Task StopAsync(CancellationToken cancellationToken) { _timer?.Change(Timeout.Infinite,0); return Task.CompletedTask; } private void DoWork(object state) { using(var scope = Services.CreateScope()) { using(var db = scope.ServiceProvider.GetRequiredService<MyDbContext>()) { ... var mq = new MessageQueue(); mq.Add("somekey", DealQueueItem); } } } private void DealQueueItem() { var mq = new MessageQueue(); var key = mq.GetCurrentKey(); var item = mq.GetItem(key); if(item!=null) { using(var scope = Services.CreateScope()) { using(var db = scope.ServiceProvider.GetRequiredService<MyDbContext>()) { //执行业务逻辑 } } } }}

当需要使用上下文的时候，首先通过IServiceProvider的CreateScope方法得到ISerivceScope，再通过它的ServiceProvider属性获取依赖倒置容器中的上下文服务。

以上，用IHostedService结合队列解决了开篇提到的场景问题，如果您有很好的想法，我们一起交流吧。文中的队列部分来自"盛派网络"的Senparc.Weixin SDK源码。