如何利用 Celery 執行定時任務並設置優先順序？

最新 06-28

本文來自作者投稿，介紹了 Celery 這個強大的 Python 第三方任務管理庫。

文 | 劉利強

Celery 是 Python 中最流行的定時任務，不僅強大而且複雜，在我的之前的一篇文章Celery 快速入門中已經介紹了如何入門使用 Celery，但是，這還很基礎，Celery 還有很多強大的功能等著我們去發掘，在我的使用過程中，我意識到有兩個功能很有必要，那就是定時任務和優先順序，本文將以個人的理解進行分享。

序言

在開始正式介紹之前，有必要說一下的是讀者最好是有一點點使用 Celery 的基礎，如果沒有的話，我之前寫過一篇入門的文章 Celery 快速入門可以快速學習一下，以便方便後面的介紹。在我使用和探索Celery的過程中，發現有兩個需求非常重要，那就是定時任務和優先順序。

首先是定時任務，這裡所謂的定時任務不僅僅是代碼寫死的定時任務，也就是說一跑起來就固定的定時任務，還有隨著代碼的運行而動態載入的定時任務。例如，舉個簡單點的例子，假如我的博客需要定時發布一篇文章，現在是下午 3 點 15 分，我希望這篇文章是晚上 8 點 17 分準時發，這裡就需要定時任務了，而且還不能是代碼寫死的，因為這是我的動態需求；

然後是一個優先順序，優先順序在很多地方都有應用，例如操作系統，各個進程運行中有個優先順序吧，重要的任務肯定要先跑，那麼對於 Celery 這樣的隊列任務肯定也需要優先順序，一些比較緊急的任務要先跑，例如用戶的簡訊驗證碼比較緊急，需要及時發送；一些宣傳的簡訊不那麼急，可以讓讓位置，緩點再發。這裡就是優先順序的作用，要分清事情的輕重緩急，才能做出更好的產品和用戶體驗，然而，後面會提到的是，Celery 其實是不支持優先順序的哦，雖然在官方文檔中聲明並非所有 Broker 都支持優先順序，只有兩個隊列 Redis 和 Beanstalk 支持，但是，經試驗和查資料，發現真不支持。

定時任務

定時任務其實和普通的非同步任務差不多，都是需要定義執行體的，所謂的執行體就是真正定時運行的代碼，所以，首先，我就先給出定義 worker 的代碼，這裡就不多做解釋了，因為代碼和 Celery 快速入門中的是一致的：

然後將 worker 運行起來，代碼都很簡單的：

然後就會很自然得看到這些輸出，說明 worker 啟動正常了：

然後先來看一下簡單的寫死的定時任務，也就是說是在代碼中固定的，隨著代碼運行的設置：

這裡的核心其實就是這裡添加了一個定時任務，就是每5秒執行一個 worker.add 任務，然後傳遞的參數是 (1, 2)，就是這麼簡單。

動態修改定時任務

然而，可以發現的是這顯然是太過於呆板了，也就是說我們的任務是固定在代碼中的，我們只能夠在代碼運行之前將它寫在代碼中，但是，並不能在運行過程中動態得更新它，有沒有什麼辦法來動態更新呢？

很遺憾，對於官方版本的 Celery 是沒有的，但是，幸運的是在 Celery 的官網介紹中有提及一下相關的信息，我們看一下：

Using django-celery『s scheduler you can add, modify and remove periodic tasks from the Django Admin.

既然這樣，我們不妨嘗試一下 Django 的任務，看下效果是怎樣的。

Django Celery Scheduler

因為我對 Django 不是很熟悉，所以我就以別人的項目為基礎，以我殘留的 Django 知識進行了些許修改，最終將項目跑起來了，然後可以看到動態添加任務的 Admin 頁面，我看到的第一眼是很震撼！！真的很震撼，非常佩服這個作者，我一定要將他開源的代碼都掃一遍，因為真的太有思想了，上一張 Admin 的圖。

這是添加定時動態任務的圖，我知道但你看這篇文章的時候肯定也想自己跑一遍，所以下面我就說下怎麼跑起來的步驟：

1. clone 項目

因為原項目只是說明性的，所以不能直接用，因此我修改後放在 github 上了，第一步你需要做的就是將它 clone 下來：

2. django 套路

將代碼拷貝下來之後，後面的工作就是套路了，第一步肯定是安裝依賴的：

然後是遷移資料庫：

接著還不能跑應用，還有一步需要做，那就是創建 admin 用戶：

後面就是輸入密碼了，隨便輸點什麼的就行了。

最後亮點來了，你需要跑兩個東西，一個是 worker，一個是網頁應用，需要注意的是，這兩個都是阻塞式的命令，所以最好打開兩個 shell 來跑最好：

運行 worker

運行 app

然後你打開瀏覽器的地址： ,輸入剛才的賬號密碼就可以了。

其他應用怎麼辦

按照 Django 的套路應該你是見識到了動態添加任務的甜頭了，可是，如果我用的不是 Django 該怎麼辦呢？總不能說特定為了動態修改任務而特地加一個 Django 用戶管理任務吧，好像有點不值得啊。

我在使用 Celery 的時候也遇到了這個問題，在查閱了一些資料之後，還是看到了官網的說明：

Custom scheduler classes can be specified on the command-line (the -S argument). The default scheduler is celery.beat.PersistentScheduler, which is simply keeping track of the last run times in a local database file (a shelve).

原來類似於 Django 的方式是自定義了一個調度器，然後我就尋找有沒有方便的調度器可以使用呢，找了好幾圈之後發現沒有滿足我需求的調度器，其實我的需求很簡單，就兩個：

支持 Redis 作為 Broker

支持動態修改任務

所以，為了自己的需求，我自己開發了一個celerybeatredis，開源在 github，下面就以 celerybeatredis 為 Scheduler 介紹一下如何動態得修改定時任務。

優先順序

談完第一個問題，我們接下來再說說第二個問題，那就是優先順序，其實就以優先順序這個概念來說，是不好實現的，因為要想完全安排任務的優先順序是有困難的，這依賴於非同步任務的調度演算法以及非同步任務的編寫，為什麼這麼說，這裡就介紹兩個場景來說明一下優先順序的問題。

關於優先順序設計的兩個場景非同步任務調度演算法

我要描述的第一個場景是關於調度器的調度演算法的，為了方便描述，我將優先順序縮小到只有兩個級別：高優先順序和低優先順序。那麼，按照我們的思維邏輯，當同時有高優先順序的任務和低優先順序的任務到來是，肯定是先執行高優先順序的任務，將低優先順序的任務放一邊。

那麼，問題是，假如我們的低優先順序和高優先順序的處理程序是同一個進程，但調度器空閑的時候，也就是低優先順序和高優先順序都沒來的時候，突然來了一個很耗時間的低優先順序任務，然後跑起來了，一會會之後，高優先順序的任務來了，但是，很不巧，我們在執行著低優先順序的，這時該怎麼辦？中斷低優先順序的？還是怎麼辦好？

針對上面的問題，也許解決方案可以是將低優先順序和高優先順序的處理進程分開，也就是說各跑一個，低優先順序的進程只執行低優先順序的，高優先順序的只執行高優先順序的，好像很合理。但是，這樣的話就不得不面對一個問題，那就是當只有一檔優先順序的任務的時候，兩個進程就只有一個很忙，一個很閑，勢必就浪費了一半的資源，這時又該怎麼辦？

非同步任務的設計

剛才說到當高優先順序和低優先順序放到同一進程會有問題，分開進程也會有問題，不知道怎麼辦。這裡放到同一個進程我們所的問題就是冗長低優先順序先運行的時候會阻塞高優先順序的任務，那麼我們就像為什麼低優先順序任務要那麼冗長？不能小一些嗎？

其實，在實際環境中低優先順序之所以放低優先順序，就是因為他們需要佔用大量的資源，並且這些任務完成的時間沒有約束，也就是我沒期望它能在一兩個小時內跑完，我允許它跑個大半晚上，這都是可能的。

在這種情況下，那我們能不能重構一些低優先順序的任務，讓它儘可能可中斷，給高優先順序的讓讓位？講道理大多數情況下是允許的，例如：

保存一下中間狀態，然後釋放出資源，等高優先順序跑完之後讀取狀態繼續，這是主動的

可以採用協程的方式，低優先順序的任務在阻塞操作的時候釋放資源等，這是被動的

但是，更理想的情況下是將耗時的任務分解成多個子任務，畢竟 Celery 還是提供 Chain 供鏈式調用的。

Celery 的處理方式

在 Celery 中，前面提高的兩個點都可以實現，因此在我們的處理過程中，使用哪種解決方式就看我們自身的需要和能力了。因為在 Celery 中本身是不支持優先順序的，但是，為什麼我說可以實現呢？那是因為 Celery 提供其他特性，可以幫助我們實現。這裡就以 Celery 的路由機製為核心，介紹一種處理優先順序的方式。

要使用優先順序，我們肯定是至少有兩個任務要競爭資源的，所以，這裡還是假設有兩個優先順序的任務，高優先順序的 add 和低優先順序的 sub，但兩個任務都過來的時候，我們需要保證的就是 add 肯定可以得到執行，sub 可能能執行，也可能不執行。

這裡使用 Celery 的解決方案就是 add 使用的是 queueA，然後 sub 使用的是 queueB，然後在啟動 Celery 的時候，我們給 queueA 啟動兩個 worker，然後給 queueB 啟動一個 worker，並且是和其中一個 queueA 共用的，這樣的話，我們就能保證 add 肯定能得到執行，sub 可能執行也可能不執行。

這裡的思想其實就是將之前討論的場景一的兩個考慮結合起來，但是，這不是最好的解決辦法，畢竟 Celery 本身是不支持優先順序的。但至少是一種折中的辦法，因為 Celery 的重點不在這。

這裡給出一個簡單的實現，我已經將代碼放到 gist(不知道為什麼，gist 需要梯子) 上，有需要嘗試的同學可以 clone 代碼嘗試一下：

然後按照你的想法修改一下 route_client.py，直接運行

Reference

Celry 快速入門

Celery Periodic Tasks

Celery Calling Tasks

Getting Started Scheduling Tasks with Celery

Setting up an asynchronous task queue for Django using Celery and Redis

利用celery+django 在admin後台設置定時任務

SOS, admin Django doesn t work!!!

celery最佳實踐