分布式任務隊列 Celery 的實踐
筆者在近期工作中有接觸到 Celery，這是一個開源的分布式任務隊列(Distributed Task Queue)，在 Github 上現有 18k star，主要可以用于實現應用中的異步任務和定時任務，雖然是用 Python 編寫，但協議可以用任何語言實現，現已有 gocelery、nodecelery 和 celery-php 等。
筆者寫下此文總結對 Celery 的了解和在工作中的使用。本文的大概內容如下：
- 任務隊列是什么;
- Celery 做了什么;
- Celery 在工作中的實踐。
任務隊列是什么
“消息隊列(Message Queue)”，后端同學應該都有了解，常見的有 RabbitMQ、RocketMQ、Kafka。而“任務隊列(Task Queue)”，筆者在接觸 Celery 之前是沒有聽過的。任務隊列是什么，而任務隊列和消息隊列，這兩者之間有何關系。帶著問題，先看看 Celery 的架構：
Celery
在 Celery 的架構中，可看出由多臺 Server 發起異步任務(Async Task)，發送任務到 Broker 的隊列中，其中的 Celery Beat 進程可負責發起定時任務。當 Task 到達 Broker 后，會將其分發給相應的 Celery Worker 進行處理。當 Task 處理完成后，其結果存儲至 Backend。
在上述過程中的 Broker 和 Backend，Celery 沒有實現，而是使用了現有開源實現，例如 RabbitMQ 作為 Broker 提供消息隊列服務，Redis 作為 Backend 提供結果存儲服務。Celery 就像是抽象了消息隊列架構中 Producer、Consumer 的實現，將消息隊列中基本單位“消息”抽象成了任務隊列中的“任務”，并將異步、定時任務的發起和結果存儲等操作進行了封裝，讓開發者可以忽略 AMQP、RabbitMQ 等實現細節，為開發帶來便利。
綜上所述，Celery 作為任務隊列是基于消息隊列的進一步封裝，其實現依賴消息隊列。
接下來，通過一個簡單的應用來具體了解 Celery 做了什么。
Celery 做了什么
在應用開發中，為了保證響應速度，耗時且不影響流程的操作通常被做異步處理。例如在用戶注冊的處理過程中，通常會異步發送郵件通知用戶，下面看看 Celery 是如何實現該異步操作。
在 task.py 中聲明了發送郵件的方法 send_mail，并為其加上 Celery 提供的 @app.task 裝飾器。通過該裝飾器，可以將 send_mail 函數變成一個 celery.app.task:Task 實例對象。而該 Task 實例可提供了兩個核心功能：
- 將消息發送給隊列;
- 聲明 Worker 接收到消息后需要執行的具體函數。
- from celery import Celery
- app = Celery('tasks', broker='amqp://guest@localhost//')
- @app.task
- def send_mail(email):
- print("send mail to ", email)
- import time
- time.sleep(5)
- return "success"
Task 已經定義完成，若要發起異步任務，可通過調用 Task 的 delay 方法，該方法會將消息發送至隊列，例如在用戶注冊完成時，發起發郵件的異步任務：
- # user.py
- from tasks import send_mail
- def register():
- print("1. 插入記錄到數據庫")
- print("2. 通過celery異步發郵件")
- send_mail.delay("chaycao@gmail.com")
- print("3. 告訴用戶注冊成功")
- if __name__ =='__main__':
- register()
運行以上程序后，消息已經發送至 RabbitMQ 的隊列中，可觀察到其消息格式如下：
Task in RabbitMQ
可看出 Celery 封裝后的消息包含了 task 標識和運行參數等內容。
接著，啟動 Worker 消費 RabbitMQ 中的消息：
- celery -A tasks worker --loglevel=info
Worker 啟動后，可以看到下面打印信息：
Worker Start
首先是 Worker 的配置信息，然后是 Worker 所執行的 Task 列表，接著是從 RabbitMQ 中成功獲取消息并執行相應的 Task。
通過以上示例，可以進一步明白 Celery 作為任務隊列框架所做的工作，而“分布式任務隊列”中的”分布式“指的則是 Producer、Consumer 可以有多個，即多個進程向 Broker 發送任務，多個 Worker 從 Broker 中獲取 Task 并執行。
以上只是一個簡單的示例，接著再看下筆者在工作中所接觸到的關于 Celery 使用的一些實踐經驗。
Celery 在工作中的實踐
根據業務場景劃分隊列
在筆者所工作的項目中，Celery 用于處理下單、解析軌跡、推送上游等異步任務和定時任務。根據每個 Task 的業務場景，可為其指定對應的隊列，例如：
- DEFAULT_CELERY_ROUTES = {
- 'celery_task.pending_create': {'queue': 'create'},
- 'celery_task.multi_create': {'queue': 'create'},
- 'celery_task.pull_tracking': {'queue': 'pull'},
- 'celery_task.pull_branch': {'queue': 'pull'},
- 'celery_task.push_tracking': {'queue': 'push'},
- 'celery_task.push_weight': {'queue': 'push'},
- }
- CELERY_ROUTES = {
- DEFAULT_CELERY_ROUTES
- }
根據業務場景，在 DEFAULT_CELERY_ROUTES 配置中指定 6 個 Task 對應的 Queue，共有 3 個隊列 create、pull、push，并將該路由規則加入到 CELERY_ROUTES 中以生效。這樣設計的目的是為了不同場景彼此之間互不影響，例如解析任務阻塞不應該影響下單任務。
進一步劃分隊列
在根據業務場景粗略劃分后，對于某個場景，可能需要更細致的劃分，例如在向上游推送時，為了避免一個上游的阻塞影響向其他上游推送，需要做到不同上游彼此之間互不影響。所以需要針對不同上游使用不同隊列，例如：
- CLIENT_CELERY_ROUTES = {
- # {0} 為 client 的占位符，在 ClientRouter 中進行格式化
- 'celery_task.push_tracking_retry': {'queue': 'push_tracking_retry_{0}'},
- 'celery_task.push_weight_retry': {'queue': 'push_weight_retry_{0}'},
- }
- class ClientRouter(object):
- def route_for_task(self, task, args=None, kwargs=None):
- if task not in CLIENT_CELERY_ROUTES:
- return None
- client_id = kwargs('client_id')
- # 根據 client_id 獲取隊列名
- queue_name = CLIENT_CELERY_ROUTES[task]['queue'].format(client_id)
- return {'queue': queue_name}
- CELERY_ROUTES = {
- 'ClientRouter'
- DEFAULT_CELERY_ROUTES,
- }
在 CLIENT_CELERY_ROUTES 中指定了需要根據 Client 隔離隊列的 Task 和其對應的 Queue 名稱格式，隊列名中含有一個占位符，為的是根據不同 Client 得到不同的隊列名。
接著實現了一個路由器 ClientRouter ，其中定義了 router_for_task 方法，其作用是為 task 指定對應的隊列名。可看出其中的邏輯是如果 task 在 CLIENT_CELERY_ROUTES 中，將會用 kwargs 中的 client_id 格式化隊列名，得到最終發送消息的隊列名，達到根據入參 client_id 來決定具體使用的隊列，從而起到隔離不同 Client 使用不同隊列的效果。
除了在 Client 的維度上劃分，若需要在其他維度進一步劃分隊列以達到隔離的效果，也可參考該方法來設計路由規則。
動態隊列
再來說說動態隊列，其本質是預備隊列，其目的是為了在線上環境減輕某些隊列消息堆積的壓力，起到快速支援的作用。通過配置來定義動態隊列需要支援哪些隊列，例如當 push 隊列的壓力較大，可配置 json 如下，將 push_tracking 和 push_weight 兩個 Task 路由到預備的動態隊列中。
- celery_dynamic_router 配置
- {
- "celery_task.push_tracking": {
- "dynamic_queue": [1,2],
- "dynamic_percentage": 0.7,
- },
- "celery_task.push_weight": {
- "dynamic_queue": [3,4],
- "dynamic_percentage": 0.7,
- }
- }
上述配置的作用是將 70% 的 celery_task.push_tracking Task 路由到動態隊列 1、2 上，70% 的 celery_task.push_weight Task 路由到動態隊列 3、4 上。
動態隊列的路由器 DynamicRouter 大致實現如下：
- class DynamicRouter(object):
- def route_for_task(self, task, args=None, kwargs=None):
- # 獲取配置
- task_config = get_conf_dict('celery_dynamic_router').get(task, None)
- # task如果沒在配置中，則直接返回
- if not task_config:
- return None
- # 獲取task對應的動態隊列配置
- dynamic_queue = task_config.get('dynamic_queue', [])
- dynamic_percentage = task_config.get('dynamic_percentage', 0.0)
- # 將一定比例的task路由到動態隊列中
- if random.random() <= dynamic_percentage:
- # 決定使用哪個動態隊列
- queue_name = router_load_balance(dynamic_queue, task_name)
- log.data('get_router| task_name:%s, queue:%s', task_name, queue_name)
- return {'queue': queue_name}
- else:
- return None
動態配置的定時任務
前文提到 Celery 不僅能實現異步任務，還能通過 Celery Beat 實現定時任務，首先看一個例子：
- from celery.schedules import crontab
- app.conf.beat_schedule = {
- # 每30秒發送一次郵件
- 'sendmail-every-30-seconds': {
- 'task': 'asks.send_mail',
- 'schedule': 30.0,
- 'args': ['chaycao@gmail.com']
- },
- }
完成上述配置后，執行 Celery Beat 命令：
celery beat
即根據配置每 30 秒執行一次 send_email 任務。
上述示例是在代碼中配置定時任務。而在筆者的工作中使用了 djcelery 提供的數據庫調度模型，通過結合 django 提供的 ORM 功能來動態設置，更為方便。下面敘述如何實現，首先在 Celery 配置中新增：
- CELERYBEAT_SCHEDULER = 'djcelery.schedulers.DatabaseScheduler'
設置使用 DatabaseScheduler，然后再生成定時任務的配置表：
- python manage.py migrate
可以看到數據庫中多出了以下表：
- | celery_taskmeta |
- | celery_tasksetmeta |
- | djcelery_crontabschedule |
- | djcelery_intervalschedule |
- | djcelery_periodictask |
- | djcelery_periodictasks |
- | djcelery_taskstate |
- | djcelery_workerstate |
完成以上操作，最后只用執行 Celery Beat 命令，則會去數據庫中讀取配置發起定時任務。這樣的好處是可以通過修改數據庫中的記錄來實現動態配置定時任務，例如調整任務的周期或者參數。
以上便是筆者在工作中接觸到 Celery 所收獲的內容，如果有需要實現異步任務、定時任務的場景，可以考慮使用 Celery。
我是草捏子，一只熱愛技術和生活的草魚，我們下期見!
參考
Message Queue vs Task Queue difference (https://newbedev.com/message-queue-vs-task-queue-difference)
高性能異步框架Celery入坑指南 (https://juejin.cn/post/6844903689103081480)
分布式任務隊列 Celery—深入 Task (https://www.cnblogs.com/jmilkfan-fanguiju/p/10589779.html)
