快捷導(dǎo)航

C#基于時間輪調(diào)度實(shí)現(xiàn)延遲任務(wù)詳解

更新時間：2022年12月31日 08:52:16 作者：a1010

在很多.net開發(fā)體系中開發(fā)者在面對調(diào)度作業(yè)需求的時候一般會選擇三方開源成熟的作業(yè)調(diào)度框架來滿足業(yè)務(wù)需求，但是有些時候可能我們只是需要一個簡易的延遲任務(wù)。本文主要分享一個簡易的基于時間輪調(diào)度的延遲任務(wù)實(shí)現(xiàn)，需要的可以參考一下

在很多.net開發(fā)體系中開發(fā)者在面對調(diào)度作業(yè)需求的時候一般會選擇三方開源成熟的作業(yè)調(diào)度框架來滿足業(yè)務(wù)需求，比如Hangfire、Quartz.NET這樣的框架。但是有些時候可能我們只是需要一個簡易的延遲任務(wù)，這個時候引入這些框架就費(fèi)力不討好了。

最簡單的粗暴的辦法當(dāng)然是：

Task.Run(async () =>
{
    //延遲xx毫秒
    await Task.Delay(time);
    //業(yè)務(wù)執(zhí)行
});

當(dāng)時作為一個開發(fā)者，有時候還是希望使用更優(yōu)雅的、可復(fù)用的一體化方案，比如可以實(shí)現(xiàn)一個簡易的時間輪來完成基于內(nèi)存的非核心重要業(yè)務(wù)的延遲調(diào)度。什么是時間輪呢，其實(shí)就是一個環(huán)形數(shù)組，每一個數(shù)組有一個插槽代表對應(yīng)時刻的任務(wù)，數(shù)組的值是一個任務(wù)隊(duì)列，假設(shè)我們有一個基于60秒的延遲時間輪，也就是說我們的任務(wù)會在不超過60秒(超過的情況增加分鐘插槽，下面會講)的情況下執(zhí)行，那么如何實(shí)現(xiàn)？下面我們將定義一段代碼來實(shí)現(xiàn)這個簡單的需求

話不多說，擼代碼，首先我們需要定義一個時間輪的Model類用于承載我們的延遲任務(wù)和任務(wù)處理器。簡單定義如下：

public class WheelTask<T>
{
    public T Data { get; set; }
    public Func<T, Task> Handle { get; set; }
}

定義很簡單，就是一個入?yún)代表要執(zhí)行的任務(wù)所需要的入?yún)ⅲ缓缶褪侨蝿?wù)的具體處理器Handle。接著我們來定義時間輪本輪的核心代碼：

可以看到時間輪其實(shí)核心就兩個東西，一個是毫秒計(jì)時器，一個是數(shù)組插槽，這里數(shù)組插槽我們使用了字典來實(shí)現(xiàn)，key值分別對應(yīng)0到59秒。每一個插槽的value對應(yīng)一個任務(wù)隊(duì)列。當(dāng)添加一個新任務(wù)的時候，輸入需要延遲的秒數(shù)，就會將任務(wù)插入到延遲多少秒對應(yīng)的插槽內(nèi)，當(dāng)計(jì)時器啟動的時候，每一跳剛好1秒，那么就會對插槽計(jì)數(shù)+1，然后去尋找當(dāng)前插槽是否有任務(wù)，有的話就會調(diào)用ExecuteTask執(zhí)行該插槽下的所有任務(wù)。

public class TimeWheel<T>
{
    int secondSlot = 0;
    DateTime wheelTime { get { return new DateTime(1, 1, 1, 0, 0, secondSlot); } }
    Dictionary<int, ConcurrentQueue<WheelTask<T>>> secondTaskQueue;
    public void Start()
    {
        new Timer(Callback, null, 0, 1000);
        secondTaskQueue = new Dictionary<int, ConcurrentQueue<WheelTask<T>>>();
        Enumerable.Range(0, 60).ToList().ForEach(x =>
        {
            secondTaskQueue.Add(x, new ConcurrentQueue<WheelTask<T>>());
        });
    }
    public async Task AddTaskAsync(int second, T data, Func<T, Task> handler)
    {
        var handTime = wheelTime.AddSeconds(second);
        if (handTime.Second != wheelTime.Second)
            secondTaskQueue[handTime.Second].Enqueue(new WheelTask<T>(data, handler));
        else
            await handler(data);
    }
    async void Callback(object o)
    {
        if (secondSlot != 59)
            secondSlot++;
        else
        {
            secondSlot = 0;
        }
        if (secondTaskQueue[secondSlot].Any())
            await ExecuteTask();
    }
    async Task ExecuteTask()
    {
        if (secondTaskQueue[secondSlot].Any())
            while (secondTaskQueue[secondSlot].Any())
                if (secondTaskQueue[secondSlot].TryDequeue(out WheelTask<T> task))
                    await task.Handle(task.Data);
    }
}

接下來就是如果我需要大于60秒的情況如何處理呢。其實(shí)就是增加分鐘插槽數(shù)組，舉個例子我有一個任務(wù)需要2分40秒后執(zhí)行，那么當(dāng)我插入到時間輪的時候我先插入到分鐘插槽，當(dāng)計(jì)時器每過去60秒，分鐘插槽值+1，當(dāng)分鐘插槽對應(yīng)有任務(wù)的時候就將這些任務(wù)從分鐘插槽里彈出再入隊(duì)到秒插槽中，這樣一個任務(wù)會先進(jìn)入插槽值=2(假設(shè)從0開始計(jì)算)的分鐘插槽，計(jì)時器運(yùn)行120秒后分鐘值從0累加到2,2插槽的任務(wù)彈出到插槽值=40的秒插槽里，當(dāng)計(jì)時器再運(yùn)行40秒，剛好就可以執(zhí)行這個延遲2分40秒的任務(wù)。話不多說，上代碼：

首先我們將任務(wù)WheelTask增加一個Second屬性，用于當(dāng)任務(wù)從分鐘插槽彈出來時需要知道自己入隊(duì)哪個秒插槽

public class WheelTask<T>
{
    ...
    public int Second { get; set; }
    ...
}

接著我們再重新定義時間輪的邏輯增加分鐘插槽值以及插槽隊(duì)列的部分

public class TimeWheel<T>
{
    int minuteSlot, secondSlot = 0;
    DateTime wheelTime { get { return new DateTime(1, 1, 1, 0, minuteSlot, secondSlot); } }
    Dictionary<int, ConcurrentQueue<WheelTask<T>>>  minuteTaskQueue, secondTaskQueue;
    public void Start()
    {
        new Timer(Callback, null, 0, 1000);、
        minuteTaskQueue = new Dictionary<int, ConcurrentQueue<WheelTask<T>>>();
        secondTaskQueue = new Dictionary<int, ConcurrentQueue<WheelTask<T>>>();
        Enumerable.Range(0, 60).ToList().ForEach(x =>
        {
            minuteTaskQueue.Add(x, new ConcurrentQueue<WheelTask<T>>());
            secondTaskQueue.Add(x, new ConcurrentQueue<WheelTask<T>>());
        });
    }
    ...
}

同樣的在添加任務(wù)的AddTaskAsync函數(shù)中我們需要增加分鐘，代碼改為這樣，當(dāng)大于1分鐘的任務(wù)會入隊(duì)到分鐘插槽中，小于1分鐘的會按原邏輯直接入隊(duì)到秒插槽中：

public async Task AddTaskAsync(int minute, int second, T data, Func<T, Task> handler)
{
    var handTime = wheelTime.AddMinutes(minute).AddSeconds(second);
        if (handTime.Minute != wheelTime.Minute)
            minuteTaskQueue[handTime.Minute].Enqueue(new WheelTask<T>(handTime.Second, data, handler));
        else
        {
            if (handTime.Second != wheelTime.Second)
                secondTaskQueue[handTime.Second].Enqueue(new WheelTask<T>(data, handler));
            else
                await handler(data);
        }
}

最后的部分就是計(jì)時器的callback以及任務(wù)執(zhí)行的部分：

async void Callback(object o)
{
    bool minuteExecuteTask = false;
    if (secondSlot != 59)
        secondSlot++;
    else
    {
        secondSlot = 0;
        minuteExecuteTask = true;
        if (minuteSlot != 59)
            minuteSlot++;
        else
        {
            minuteSlot = 0;
        }
    }
    if (minuteExecuteTask || secondTaskQueue[secondSlot].Any())
        await ExecuteTask(minuteExecuteTask);
}
async Task ExecuteTask(bool minuteExecuteTask)
{
    if (minuteExecuteTask)
        while (minuteTaskQueue[minuteSlot].Any())
            if (minuteTaskQueue[minuteSlot].TryDequeue(out WheelTask<T> task))
                secondTaskQueue[task.Second].Enqueue(task);
    if (secondTaskQueue[secondSlot].Any())
        while (secondTaskQueue[secondSlot].Any())
            if (secondTaskQueue[secondSlot].TryDequeue(out WheelTask<T> task))
                await task.Handle(task.Data);
}

基本上基于分鐘+秒的時間輪延遲任務(wù)核心功能就這些了，聰明的你一定知道如何擴(kuò)展增加小時，天，月份甚至年份的時間輪了。雖然從代碼邏輯上可以實(shí)現(xiàn)，但是大部分情況下我們使用時間輪僅僅是完成一些內(nèi)存易失性的非核心的任務(wù)延遲調(diào)度，實(shí)現(xiàn)天，周，月年意義不是很大。所以基本上到小時就差不多了。再多就上作業(yè)系統(tǒng)來調(diào)度吧。

到此這篇關(guān)于C#基于時間輪調(diào)度實(shí)現(xiàn)延遲任務(wù)詳解的文章就介紹到這了,更多相關(guān)C#延遲任務(wù)內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: