級(jí)Celery定時(shí)任務(wù)體系搭建實(shí)戰(zhàn))
1. 從一次線上事故說起為什么我最終選了Celery做定時(shí)任務(wù)去年冬天的一個(gè)凌晨我被報(bào)警電話吵醒。一個(gè)跑了半年的數(shù)據(jù)同步腳本突然不執(zhí)行了導(dǎo)致下游三個(gè)業(yè)務(wù)模塊的數(shù)據(jù)全部延遲了六個(gè)小時(shí)。排查下來原因很樸素那臺(tái)服務(wù)器上跑著七八個(gè)用系統(tǒng)crontab管理的定時(shí)腳本其中一個(gè)腳本因?yàn)橐蕾嚨腜ython包版本沖突把整個(gè)crontab的執(zhí)行環(huán)境搞崩了其他腳本跟著一起遭殃。這件事之后我開始認(rèn)真思考一個(gè)問題當(dāng)定時(shí)任務(wù)從三五個(gè)變成幾十個(gè)從單機(jī)變成多機(jī)從“跑完就行”變成“必須知道跑沒跑、跑了多久、失敗了怎么辦”的時(shí)候系統(tǒng)自帶的crontab還夠用嗎答案顯然是不夠的。我需要的是一個(gè)能統(tǒng)一管理、能追蹤狀態(tài)、能自動(dòng)重試、能水平擴(kuò)展的定時(shí)任務(wù)方案。在對比了幾種常見做法之后我最終選擇了Celery。這篇文章就把我從零搭建Celery定時(shí)任務(wù)體系的完整過程拆開來講包括架構(gòu)設(shè)計(jì)的取舍、Beat調(diào)度器的配置細(xì)節(jié)、任務(wù)冪等性的處理、以及那些只有真正跑在生產(chǎn)環(huán)境才會(huì)遇到的坑。如果你正在被散落各處的定時(shí)腳本折磨或者團(tuán)隊(duì)里沒人說得清“到底有多少個(gè)定時(shí)任務(wù)在跑”那這篇內(nèi)容應(yīng)該能幫你少走不少彎路。我會(huì)盡量把每個(gè)決策背后的“為什么”講清楚而不是只丟一堆配置讓你抄。2. Celery定時(shí)任務(wù)的核心組件與運(yùn)行邏輯2.1 三個(gè)角色各干什么活Celery做定時(shí)任務(wù)核心就三個(gè)角色Beat、Worker和Broker。很多人一開始搞混它們的關(guān)系我用一個(gè)生活化的類比來解釋。想象你開了一家餐廳。Beat就是那個(gè)每天早上列好“今天幾點(diǎn)該做什么菜”的排班經(jīng)理它不炒菜只負(fù)責(zé)到點(diǎn)喊人。Broker是傳菜窗口經(jīng)理把訂單寫在小票上放到窗口誰有空誰就來取。Worker就是后廚的廚師看到窗口有小票就取走按照票上的要求把菜做出來。這個(gè)類比里最關(guān)鍵的一點(diǎn)是Beat只負(fù)責(zé)“發(fā)號(hào)施令”真正干活的是Worker。所以如果你的定時(shí)任務(wù)沒執(zhí)行首先要確認(rèn)的是Beat有沒有正常發(fā)出任務(wù)而不是一上來就查Worker。具體到技術(shù)層面Beat進(jìn)程會(huì)根據(jù)配置的調(diào)度規(guī)則在指定時(shí)間把任務(wù)消息發(fā)送到Broker通常是Redis或RabbitMQ。Worker進(jìn)程持續(xù)監(jiān)聽Broker一旦發(fā)現(xiàn)有新任務(wù)就取出來執(zhí)行。執(zhí)行結(jié)果可以選擇性地存到Result Backend里方便后續(xù)查詢?nèi)蝿?wù)狀態(tài)。這里有個(gè)容易踩的坑Beat和Worker必須是兩個(gè)獨(dú)立的進(jìn)程。我見過有同事把Beat和Worker寫在同一個(gè)啟動(dòng)腳本里結(jié)果Worker一重啟Beat也跟著斷了所有定時(shí)任務(wù)全部停擺。正確的做法是用兩個(gè)獨(dú)立的進(jìn)程管理工具分別守護(hù)它們。2.2 調(diào)度規(guī)則是怎么被解析的Celery的定時(shí)調(diào)度規(guī)則寫在beat_schedule配置里每條規(guī)則本質(zhì)上是一個(gè)字典包含任務(wù)名、任務(wù)路徑、調(diào)度周期和參數(shù)。調(diào)度周期的表達(dá)方式有好幾種最常用的是crontab和timedelta。crontab適合“每天幾點(diǎn)幾分執(zhí)行”這種固定時(shí)刻的場景。比如每天早上八點(diǎn)半跑一次數(shù)據(jù)匯總from celery.schedules import crontab app.conf.beat_schedule { morning-data-summary: { task: tasks.summarize_daily_data, schedule: crontab(hour8, minute30), }, }timedelta適合“每隔多久執(zhí)行一次”的場景。比如每十分鐘檢查一次隊(duì)列積壓from datetime import timedelta app.conf.beat_schedule { check-queue-every-10min: { task: tasks.check_queue_backlog, schedule: timedelta(minutes10), }, }這兩種方式的選擇邏輯很簡單如果任務(wù)必須在某個(gè)絕對時(shí)刻執(zhí)行比如財(cái)務(wù)報(bào)表要在每天零點(diǎn)生成用crontab如果任務(wù)只關(guān)心間隔比如心跳檢測用timedelta?；煊靡矝]問題同一個(gè)beat_schedule里可以同時(shí)存在兩種規(guī)則。2.3 任務(wù)消息在Broker里長什么樣理解任務(wù)消息的結(jié)構(gòu)對排查問題非常有幫助。當(dāng)Beat觸發(fā)一個(gè)任務(wù)時(shí)它實(shí)際上往Broker里塞了一條JSON格式的消息大致長這樣{ task: tasks.summarize_daily_data, id: a1b2c3d4-e5f6-7890-abcd-ef1234567890, args: [], kwargs: {date: 2024-01-15}, retries: 0, eta: null, expires: null }這條消息里的id是任務(wù)的唯一標(biāo)識(shí)后面查任務(wù)狀態(tài)就靠它。retries記錄當(dāng)前重試次數(shù)Celery會(huì)根據(jù)配置的max_retries決定還要不要繼續(xù)重試。expires是一個(gè)很有用的字段可以設(shè)置任務(wù)的過期時(shí)間超過這個(gè)時(shí)間還沒被執(zhí)行就直接丟棄避免堆積一堆過期的任務(wù)把Worker壓垮。我一般會(huì)在任務(wù)配置里加上expires比如數(shù)據(jù)同步任務(wù)設(shè)置兩小時(shí)過期。這樣即使Worker因?yàn)槟撤N原因堵住了恢復(fù)之后也不會(huì)去執(zhí)行那些早就沒有意義的舊任務(wù)。3. 環(huán)境搭建從零到跑通第一個(gè)定時(shí)任務(wù)3.1 依賴安裝與Broker選型先把基礎(chǔ)依賴裝上。Celery本身很輕量核心包就一個(gè)pip install celery但光有Celery還不夠你需要一個(gè)Broker。常見的選擇是Redis和RabbitMQ我兩個(gè)都用過說說各自的適用場景。Redis的優(yōu)勢是部署簡單、性能好、大家普遍熟悉。如果你團(tuán)隊(duì)已經(jīng)在用Redis做緩存直接復(fù)用同一個(gè)實(shí)例用不同的db編號(hào)隔離是最省事的方案。缺點(diǎn)是Redis作為Broker時(shí)消息的可靠性保障不如RabbitMQ極端情況下比如Redis突然宕機(jī)且沒有持久化可能丟消息。RabbitMQ的優(yōu)勢是消息可靠性更強(qiáng)支持更復(fù)雜的路由策略適合對任務(wù)丟失零容忍的場景。缺點(diǎn)是運(yùn)維成本更高需要單獨(dú)維護(hù)一個(gè)RabbitMQ集群。我的建議是如果定時(shí)任務(wù)的結(jié)果不是“丟了就出大事”的那種用Redis就夠了如果是涉及資金、訂單等關(guān)鍵業(yè)務(wù)老老實(shí)實(shí)上RabbitMQ。我現(xiàn)在的項(xiàng)目用的是Redis因?yàn)槎〞r(shí)任務(wù)主要是數(shù)據(jù)同步和報(bào)表生成偶爾丟一次可以手動(dòng)補(bǔ)跑。安裝Redis的Python客戶端pip install redis3.2 Celery應(yīng)用的初始化配置接下來創(chuàng)建一個(gè)Celery應(yīng)用實(shí)例。我習(xí)慣把配置拆成獨(dú)立的文件方便不同環(huán)境開發(fā)、測試、生產(chǎn)用不同的參數(shù)。先建一個(gè)celery_app.pyfrom celery import Celery app Celery(myproject) app.config_from_object(celery_config) app.autodiscover_tasks([tasks])然后建一個(gè)celery_config.pybroker_url redis://:password127.0.0.1:6379/0 result_backend redis://:password127.0.0.1:6379/1 timezone Asia/Shanghai enable_utc False task_serializer json result_serializer json accept_content [json] task_track_started True task_time_limit 3600 task_soft_time_limit 3000 worker_max_tasks_per_child 200 worker_prefetch_multiplier 1這里有幾個(gè)配置值得展開說。timezone和enable_utc必須一起設(shè)置。Celery默認(rèn)用UTC時(shí)間如果你不改成東八區(qū)你寫的crontab(hour8)實(shí)際上會(huì)在北京時(shí)間下午四點(diǎn)執(zhí)行。這個(gè)坑我踩過當(dāng)時(shí)納悶了一整天為什么任務(wù)不在早上跑。task_track_started設(shè)為True之后任務(wù)開始執(zhí)行時(shí)狀態(tài)會(huì)變成STARTED方便你區(qū)分“排隊(duì)中”和“正在跑”。task_time_limit是硬超時(shí)到時(shí)間直接殺掉Worker進(jìn)程task_soft_time_limit是軟超時(shí)會(huì)拋出一個(gè)異常讓任務(wù)自己處理。我一般設(shè)軟超時(shí)比硬超時(shí)少幾分鐘給任務(wù)一個(gè)優(yōu)雅退出的機(jī)會(huì)。worker_prefetch_multiplier設(shè)為1很重要。默認(rèn)值是4意味著每個(gè)Worker會(huì)一次性從Broker預(yù)取4個(gè)任務(wù)。對于執(zhí)行時(shí)間較長的定時(shí)任務(wù)來說這會(huì)導(dǎo)致任務(wù)分配不均——一個(gè)Worker手里壓著好幾個(gè)任務(wù)慢慢做其他Worker閑著。設(shè)為1就是“做完一個(gè)再拿一個(gè)”。3.3 寫出第一個(gè)可用的定時(shí)任務(wù)配置好了之后寫一個(gè)最簡單的任務(wù)來驗(yàn)證鏈路是否通暢。在tasks/目錄下建一個(gè)demo.pyfrom celery_app import app import logging logger logging.getLogger(__name__) app.task(bindTrue, max_retries3, default_retry_delay60) def heartbeat(self): logger.info(心跳任務(wù)執(zhí)行中任務(wù)ID: %s, self.request.id) return ok然后在celery_config.py里注冊調(diào)度規(guī)則beat_schedule { heartbeat-every-minute: { task: tasks.demo.heartbeat, schedule: 60.0, }, }啟動(dòng)Beat和Worker開兩個(gè)終端# 終端1啟動(dòng)Beat celery -A celery_app beat -l info # 終端2啟動(dòng)Worker celery -A celery_app worker -l info -Q celery如果一切正常你會(huì)在Beat的日志里看到它每分鐘往Broker發(fā)一條任務(wù)消息在Worker的日志里看到任務(wù)被執(zhí)行并返回了結(jié)果。注意Beat進(jìn)程默認(rèn)會(huì)把調(diào)度狀態(tài)存到一個(gè)本地文件里celerybeat-schedule。如果你在多臺(tái)機(jī)器上同時(shí)啟動(dòng)Beat它們會(huì)各自維護(hù)自己的調(diào)度狀態(tài)導(dǎo)致任務(wù)重復(fù)觸發(fā)。Beat必須是單點(diǎn)的這一點(diǎn)后面還會(huì)展開講。4. 生產(chǎn)環(huán)境才會(huì)暴露的五個(gè)坑4.1 Beat單點(diǎn)問題與高可用方案前面提到Beat必須是單點(diǎn)但單點(diǎn)意味著如果Beat掛了所有定時(shí)任務(wù)都會(huì)停。這個(gè)問題在生產(chǎn)環(huán)境是致命的。我試過幾種方案。最簡單的是用supervisor守護(hù)Beat進(jìn)程掛了自動(dòng)重啟。但這只能解決進(jìn)程崩潰的情況如果服務(wù)器本身宕機(jī)還是會(huì)有空窗期。更可靠的方案是用分布式鎖保證同一時(shí)刻只有一個(gè)Beat在運(yùn)行。具體做法是啟動(dòng)多個(gè)Beat實(shí)例但每個(gè)實(shí)例在觸發(fā)任務(wù)之前先搶一把Redis鎖搶到鎖的才真正發(fā)送任務(wù)。Celery本身沒有內(nèi)置這個(gè)機(jī)制需要自己實(shí)現(xiàn)一個(gè)自定義的Scheduler類。from celery.beat import Scheduler import redis class LockedScheduler(Scheduler): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.redis redis.Redis(host127.0.0.1, port6379, db2) def tick(self): lock self.redis.lock(celery_beat_lock, timeout30) if lock.acquire(blockingFalse): try: return super().tick() finally: lock.release() return 0.0然后在啟動(dòng)Beat時(shí)指定這個(gè)Schedulercelery -A celery_app beat -S celery_app.LockedScheduler -l info這樣即使起了多個(gè)Beat同一時(shí)刻也只有一個(gè)在真正工作。不過說實(shí)話這個(gè)方案增加了復(fù)雜度如果你的定時(shí)任務(wù)不是特別關(guān)鍵用supervisor守護(hù)單點(diǎn)Beat 監(jiān)控告警也夠用了。4.2 任務(wù)重復(fù)執(zhí)行的冪等性處理定時(shí)任務(wù)最怕的就是重復(fù)執(zhí)行。網(wǎng)絡(luò)抖動(dòng)、Worker超時(shí)重啟、Beat重復(fù)觸發(fā)都可能導(dǎo)致同一個(gè)任務(wù)被執(zhí)行兩次。如果你的任務(wù)是“給用戶發(fā)優(yōu)惠券”重復(fù)執(zhí)行就是事故。解決思路是讓任務(wù)具備冪等性——執(zhí)行一次和執(zhí)行多次的結(jié)果是一樣的。具體做法有幾種第一種是用數(shù)據(jù)庫唯一約束。比如插入數(shù)據(jù)時(shí)用INSERT IGNORE或ON CONFLICT DO NOTHING重復(fù)執(zhí)行不會(huì)產(chǎn)生臟數(shù)據(jù)。第二種是在任務(wù)開頭檢查狀態(tài)。比如app.task def sync_order_data(date_str): if cache.get(fsync_done:{date_str}): logger.info(任務(wù)已執(zhí)行過跳過) return # 執(zhí)行同步邏輯 do_sync(date_str) cache.set(fsync_done:{date_str}, 1, ex86400)第三種是用Celery的task_id做去重。在Beat配置里給每個(gè)任務(wù)指定固定的task_idWorker端維護(hù)一個(gè)已執(zhí)行task_id的集合遇到重復(fù)的直接跳過。我一般用第二種簡單直接而且緩存過期時(shí)間設(shè)長一點(diǎn)比如24小時(shí)就能覆蓋絕大多數(shù)重復(fù)場景。4.3 Worker內(nèi)存泄漏與自動(dòng)重啟Python的垃圾回收機(jī)制在處理循環(huán)引用時(shí)有時(shí)候不夠及時(shí)長時(shí)間運(yùn)行的Worker進(jìn)程內(nèi)存會(huì)慢慢漲上去。我遇到過一次Worker跑了三天之后內(nèi)存占用從200MB漲到了2GB最后被系統(tǒng)OOM Killer干掉了。Celery提供了一個(gè)配置來解決這個(gè)問題worker_max_tasks_per_child 200這個(gè)配置的意思是Worker每執(zhí)行200個(gè)任務(wù)就自動(dòng)重啟一次子進(jìn)程釋放內(nèi)存。200這個(gè)數(shù)字可以根據(jù)你的任務(wù)內(nèi)存占用情況調(diào)整。如果單個(gè)任務(wù)內(nèi)存占用大就調(diào)小一點(diǎn)如果任務(wù)很輕量可以調(diào)大。還有一個(gè)相關(guān)的配置是worker_max_memory_per_child單位是KB。比如設(shè)為200000就是200MB子進(jìn)程內(nèi)存超過這個(gè)值就重啟。這個(gè)更精確但需要你對任務(wù)的內(nèi)存占用有大致了解。4.4 任務(wù)超時(shí)后的狀態(tài)不一致前面說了task_time_limit和task_soft_time_limit但這里有個(gè)隱蔽的問題硬超時(shí)殺掉Worker之后任務(wù)在Broker里的狀態(tài)可能還是“正在執(zhí)行”。如果你依賴Result Backend查任務(wù)狀態(tài)會(huì)看到任務(wù)永遠(yuǎn)卡在STARTED。解決辦法是在任務(wù)代碼里捕獲軟超時(shí)異常主動(dòng)更新狀態(tài)from celery.exceptions import SoftTimeLimitExceeded app.task(bindTrue, soft_time_limit3000, time_limit3600) def long_running_task(self): try: do_heavy_work() except SoftTimeLimitExceeded: self.update_state(stateFAILURE, meta{error: 任務(wù)超時(shí)}) raise這樣即使任務(wù)被強(qiáng)制終止?fàn)顟B(tài)也能被正確標(biāo)記。另外建議給Result Backend設(shè)置過期時(shí)間避免歷史任務(wù)狀態(tài)無限堆積result_expires 864004.5 時(shí)鐘漂移導(dǎo)致的任務(wù)偏移這個(gè)問題比較隱蔽。如果你的服務(wù)器時(shí)鐘和標(biāo)準(zhǔn)時(shí)間有偏差Beat觸發(fā)任務(wù)的時(shí)間就會(huì)不準(zhǔn)。我遇到過一臺(tái)服務(wù)器的時(shí)鐘慢了五分鐘導(dǎo)致所有定時(shí)任務(wù)都延遲了五分鐘執(zhí)行。解決辦法是配置NTP同步。在Linux上sudo apt install ntpdate sudo ntpdate pool.ntp.org或者用systemd的timesyncdsudo timedatectl set-ntp true另外Celery本身也有一個(gè)beat_max_loop_interval配置控制Beat檢查調(diào)度規(guī)則的最大間隔。默認(rèn)是300秒意味著Beat最多每5分鐘檢查一次有沒有任務(wù)要觸發(fā)。如果你的定時(shí)任務(wù)精度要求到分鐘級(jí)建議把這個(gè)值調(diào)小beat_max_loop_interval 605. 監(jiān)控與告警讓定時(shí)任務(wù)不再“黑盒”5.1 用Flower實(shí)時(shí)查看任務(wù)狀態(tài)Celery生態(tài)里最好用的監(jiān)控工具是Flower。它是一個(gè)Web界面能實(shí)時(shí)看到Worker狀態(tài)、任務(wù)執(zhí)行歷史、隊(duì)列長度等信息。安裝和啟動(dòng)pip install flower celery -A celery_app flower --port5555啟動(dòng)后訪問http://localhost:5555就能看到儀表盤。我一般會(huì)關(guān)注幾個(gè)指標(biāo)隊(duì)列積壓數(shù)量如果持續(xù)增長說明Worker不夠用、任務(wù)平均執(zhí)行時(shí)間突然變長可能是數(shù)據(jù)量增大或代碼有問題、失敗任務(wù)列表及時(shí)發(fā)現(xiàn)問題。Flower還支持配置認(rèn)證生產(chǎn)環(huán)境記得加上celery -A celery_app flower --basic_authadmin:password5.2 任務(wù)失敗后的告警接入光有監(jiān)控界面還不夠沒人會(huì)一直盯著屏幕看。需要配置告警任務(wù)失敗時(shí)主動(dòng)通知。Celery提供了任務(wù)失敗的回調(diào)鉤子可以在配置里指定def on_task_failure(self, exc, task_id, args, kwargs, einfo): send_alert(f任務(wù) {self.name} 執(zhí)行失敗任務(wù)ID: {task_id}錯(cuò)誤: {exc}) app.conf.task_annotations { *: {on_failure: on_task_failure} }send_alert可以對接郵件、企業(yè)微信機(jī)器人、釘釘機(jī)器人等。我一般用企業(yè)微信機(jī)器人簡單可靠import requests def send_alert(message): webhook https://qyapi.weixin.qq.com/cgi-bin/webhook/send?keyyour_key requests.post(webhook, json{ msgtype: text, text: {content: message} })注意告警要設(shè)置頻率限制避免任務(wù)大面積失敗時(shí)被告警轟炸。可以在send_alert里加一個(gè)簡單的計(jì)數(shù)器同一任務(wù)短時(shí)間內(nèi)失敗多次只發(fā)一條告警。5.3 任務(wù)執(zhí)行日志的集中管理Celery的日志默認(rèn)輸出到stdout在生產(chǎn)環(huán)境需要收集起來方便排查。我一般用兩種方式結(jié)合一是把Worker日志寫到文件用logrotate做切割celery -A celery_app worker -l info --logfile/var/log/celery/worker.log二是任務(wù)內(nèi)部的關(guān)鍵日志直接寫到數(shù)據(jù)庫或日志服務(wù)方便按任務(wù)ID檢索。比如app.task(bindTrue) def process_data(self, date_str): logger.info(開始處理 %s 的數(shù)據(jù)任務(wù)ID: %s, date_str, self.request.id) try: result do_process(date_str) logger.info(處理完成結(jié)果: %s, result) except Exception as e: logger.error(處理失敗: %s, e, exc_infoTrue) raise這樣出問題的時(shí)候拿著任務(wù)ID就能在日志系統(tǒng)里搜到完整的執(zhí)行鏈路。6. 性能調(diào)優(yōu)與擴(kuò)展思路6.1 Worker并發(fā)數(shù)的合理設(shè)置Worker的并發(fā)數(shù)由--concurrency參數(shù)控制默認(rèn)是CPU核心數(shù)。但這個(gè)默認(rèn)值不一定適合你的場景。如果任務(wù)是CPU密集型的比如圖像處理、大量計(jì)算并發(fā)數(shù)設(shè)為CPU核心數(shù)或核心數(shù)1比較合適再多會(huì)因?yàn)樯舷挛那袚Q反而變慢。如果任務(wù)是IO密集型的比如網(wǎng)絡(luò)請求、數(shù)據(jù)庫讀寫并發(fā)數(shù)可以設(shè)大很多因?yàn)閃orker大部分時(shí)間在等IO。我一般設(shè)成CPU核心數(shù)的2到4倍。celery -A celery_app worker -l info --concurrency8另外如果你的任務(wù)對內(nèi)存敏感可以用--poolprefork默認(rèn)配合worker_max_tasks_per_child來控制。如果任務(wù)非常輕量且數(shù)量巨大可以考慮--poolgevent或--pooleventlet用協(xié)程提高并發(fā)。6.2 任務(wù)路由與隊(duì)列隔離當(dāng)定時(shí)任務(wù)種類變多之后不同類型的任務(wù)混在同一個(gè)隊(duì)列里會(huì)互相影響。比如一個(gè)耗時(shí)的報(bào)表生成任務(wù)堵住了隊(duì)列后面所有輕量的心跳檢測任務(wù)都跟著延遲。解決辦法是用隊(duì)列隔離。在配置里定義多個(gè)隊(duì)列task_routes { tasks.report.*: {queue: report}, tasks.sync.*: {queue: sync}, tasks.heartbeat: {queue: default}, }然后啟動(dòng)不同的Worker監(jiān)聽不同的隊(duì)列celery -A celery_app worker -Q report -c 2 -l info celery -A celery_app worker -Q sync -c 4 -l info celery -A celery_app worker -Q default -c 2 -l info這樣報(bào)表任務(wù)再慢也不會(huì)影響心跳檢測。代價(jià)是需要維護(hù)更多的Worker進(jìn)程根據(jù)實(shí)際情況權(quán)衡。6.3 大規(guī)模任務(wù)下的Beat性能當(dāng)beat_schedule里的規(guī)則超過幾百條時(shí)Beat的調(diào)度性能會(huì)開始下降。因?yàn)锽eat是單線程的每次tick都要遍歷所有規(guī)則判斷是否該觸發(fā)。優(yōu)化思路有幾個(gè)。一是合并同類任務(wù)比如把“每分鐘檢查一次A”和“每分鐘檢查一次B”合并成一個(gè)任務(wù)內(nèi)部依次執(zhí)行。二是減少不必要的規(guī)則定期清理已經(jīng)廢棄的定時(shí)任務(wù)。三是用自定義Scheduler做分片把規(guī)則分散到多個(gè)Beat實(shí)例上每個(gè)實(shí)例只負(fù)責(zé)一部分規(guī)則。我目前的項(xiàng)目有大概80條調(diào)度規(guī)則Beat運(yùn)行很穩(wěn)定CPU占用可以忽略不計(jì)。所以除非你的規(guī)則數(shù)量真的很大否則不用太擔(dān)心這個(gè)問題。7. 一些零散但重要的經(jīng)驗(yàn)關(guān)于任務(wù)參數(shù)的設(shè)計(jì)我建議盡量傳簡單的、可序列化的參數(shù)。Celery默認(rèn)用JSON序列化傳不了復(fù)雜的Python對象。如果需要傳復(fù)雜數(shù)據(jù)傳一個(gè)ID讓任務(wù)自己去數(shù)據(jù)庫查。關(guān)于時(shí)區(qū)再強(qiáng)調(diào)一次timezone和enable_utc一定要配對設(shè)置。我見過太多人只設(shè)了timezone忘了enable_utc結(jié)果任務(wù)執(zhí)行時(shí)間怎么都對不上。關(guān)于Beat的調(diào)度文件celerybeat-schedule這個(gè)文件不要隨便刪。它記錄了上次調(diào)度的時(shí)間戳刪掉之后Beat會(huì)重新計(jì)算可能導(dǎo)致任務(wù)重復(fù)觸發(fā)。如果確實(shí)需要重置先停掉Beat再刪。關(guān)于任務(wù)命名建議用完整的模塊路徑比如tasks.report.generate_daily不要用簡寫。這樣在Flower里排查問題時(shí)能一眼看出任務(wù)在哪個(gè)文件里。關(guān)于測試Celery任務(wù)可以用task_always_eager配置在本地同步執(zhí)行方便調(diào)試task_always_eager True # 僅開發(fā)環(huán)境使用這樣調(diào)用task.delay()會(huì)直接在當(dāng)前進(jìn)程執(zhí)行不走Broker斷點(diǎn)調(diào)試很方便。但記得生產(chǎn)環(huán)境一定要關(guān)掉。最后說一個(gè)我踩過的坑不要在任務(wù)函數(shù)里直接操作全局變量或數(shù)據(jù)庫連接。Celery的Worker是多進(jìn)程模型每個(gè)子進(jìn)程有獨(dú)立的內(nèi)存空間全局變量在子進(jìn)程之間不共享。數(shù)據(jù)庫連接也要用連接池或者在任務(wù)內(nèi)部創(chuàng)建和關(guān)閉避免連接泄漏。這套Celery定時(shí)任務(wù)體系我已經(jīng)跑了快一年管理著六十多個(gè)定時(shí)任務(wù)日均執(zhí)行幾千次整體很穩(wěn)定。中間遇到過的問題基本都在這篇文章里了希望對正在搭建類似系統(tǒng)的你有幫助。如果只能記住一件事那就是Beat單點(diǎn)、任務(wù)冪等、失敗告警這三點(diǎn)做好了定時(shí)任務(wù)就不會(huì)出大問題。 從線上事故到生產(chǎn)級(jí)Celery定時(shí)任務(wù)體系搭建實(shí)戰(zhàn)
1. 從一次線上事故說起為什么我最終選了Celery做定時(shí)任務(wù)去年冬天的一個(gè)凌晨我被報(bào)警電話吵醒。一個(gè)跑了半年的數(shù)據(jù)同步腳本突然不執(zhí)行了導(dǎo)致下游三個(gè)業(yè)務(wù)模塊的數(shù)據(jù)全部延遲了六個(gè)小時(shí)。排查下來原因很樸素那臺(tái)服務(wù)器上跑著七八個(gè)用系統(tǒng)crontab管理的定時(shí)腳本其中一個(gè)腳本因?yàn)橐蕾嚨腜ython包版本沖突把整個(gè)crontab的執(zhí)行環(huán)境搞崩了其他腳本跟著一起遭殃。這件事之后我開始認(rèn)真思考一個(gè)問題當(dāng)定時(shí)任務(wù)從三五個(gè)變成幾十個(gè)從單機(jī)變成多機(jī)從“跑完就行”變成“必須知道跑沒跑、跑了多久、失敗了怎么辦”的時(shí)候系統(tǒng)自帶的crontab還夠用嗎答案顯然是不夠的。我需要的是一個(gè)能統(tǒng)一管理、能追蹤狀態(tài)、能自動(dòng)重試、能水平擴(kuò)展的定時(shí)任務(wù)方案。在對比了幾種常見做法之后我最終選擇了Celery。這篇文章就把我從零搭建Celery定時(shí)任務(wù)體系的完整過程拆開來講包括架構(gòu)設(shè)計(jì)的取舍、Beat調(diào)度器的配置細(xì)節(jié)、任務(wù)冪等性的處理、以及那些只有真正跑在生產(chǎn)環(huán)境才會(huì)遇到的坑。如果你正在被散落各處的定時(shí)腳本折磨或者團(tuán)隊(duì)里沒人說得清“到底有多少個(gè)定時(shí)任務(wù)在跑”那這篇內(nèi)容應(yīng)該能幫你少走不少彎路。我會(huì)盡量把每個(gè)決策背后的“為什么”講清楚而不是只丟一堆配置讓你抄。2. Celery定時(shí)任務(wù)的核心組件與運(yùn)行邏輯2.1 三個(gè)角色各干什么活Celery做定時(shí)任務(wù)核心就三個(gè)角色Beat、Worker和Broker。很多人一開始搞混它們的關(guān)系我用一個(gè)生活化的類比來解釋。想象你開了一家餐廳。Beat就是那個(gè)每天早上列好“今天幾點(diǎn)該做什么菜”的排班經(jīng)理它不炒菜只負(fù)責(zé)到點(diǎn)喊人。Broker是傳菜窗口經(jīng)理把訂單寫在小票上放到窗口誰有空誰就來取。Worker就是后廚的廚師看到窗口有小票就取走按照票上的要求把菜做出來。這個(gè)類比里最關(guān)鍵的一點(diǎn)是Beat只負(fù)責(zé)“發(fā)號(hào)施令”真正干活的是Worker。所以如果你的定時(shí)任務(wù)沒執(zhí)行首先要確認(rèn)的是Beat有沒有正常發(fā)出任務(wù)而不是一上來就查Worker。具體到技術(shù)層面Beat進(jìn)程會(huì)根據(jù)配置的調(diào)度規(guī)則在指定時(shí)間把任務(wù)消息發(fā)送到Broker通常是Redis或RabbitMQ。Worker進(jìn)程持續(xù)監(jiān)聽Broker一旦發(fā)現(xiàn)有新任務(wù)就取出來執(zhí)行。執(zhí)行結(jié)果可以選擇性地存到Result Backend里方便后續(xù)查詢?nèi)蝿?wù)狀態(tài)。這里有個(gè)容易踩的坑Beat和Worker必須是兩個(gè)獨(dú)立的進(jìn)程。我見過有同事把Beat和Worker寫在同一個(gè)啟動(dòng)腳本里結(jié)果Worker一重啟Beat也跟著斷了所有定時(shí)任務(wù)全部停擺。正確的做法是用兩個(gè)獨(dú)立的進(jìn)程管理工具分別守護(hù)它們。2.2 調(diào)度規(guī)則是怎么被解析的Celery的定時(shí)調(diào)度規(guī)則寫在beat_schedule配置里每條規(guī)則本質(zhì)上是一個(gè)字典包含任務(wù)名、任務(wù)路徑、調(diào)度周期和參數(shù)。調(diào)度周期的表達(dá)方式有好幾種最常用的是crontab和timedelta。crontab適合“每天幾點(diǎn)幾分執(zhí)行”這種固定時(shí)刻的場景。比如每天早上八點(diǎn)半跑一次數(shù)據(jù)匯總from celery.schedules import crontab app.conf.beat_schedule { morning-data-summary: { task: tasks.summarize_daily_data, schedule: crontab(hour8, minute30), }, }timedelta適合“每隔多久執(zhí)行一次”的場景。比如每十分鐘檢查一次隊(duì)列積壓from datetime import timedelta app.conf.beat_schedule { check-queue-every-10min: { task: tasks.check_queue_backlog, schedule: timedelta(minutes10), }, }這兩種方式的選擇邏輯很簡單如果任務(wù)必須在某個(gè)絕對時(shí)刻執(zhí)行比如財(cái)務(wù)報(bào)表要在每天零點(diǎn)生成用crontab如果任務(wù)只關(guān)心間隔比如心跳檢測用timedelta?；煊靡矝]問題同一個(gè)beat_schedule里可以同時(shí)存在兩種規(guī)則。2.3 任務(wù)消息在Broker里長什么樣理解任務(wù)消息的結(jié)構(gòu)對排查問題非常有幫助。當(dāng)Beat觸發(fā)一個(gè)任務(wù)時(shí)它實(shí)際上往Broker里塞了一條JSON格式的消息大致長這樣{ task: tasks.summarize_daily_data, id: a1b2c3d4-e5f6-7890-abcd-ef1234567890, args: [], kwargs: {date: 2024-01-15}, retries: 0, eta: null, expires: null }這條消息里的id是任務(wù)的唯一標(biāo)識(shí)后面查任務(wù)狀態(tài)就靠它。retries記錄當(dāng)前重試次數(shù)Celery會(huì)根據(jù)配置的max_retries決定還要不要繼續(xù)重試。expires是一個(gè)很有用的字段可以設(shè)置任務(wù)的過期時(shí)間超過這個(gè)時(shí)間還沒被執(zhí)行就直接丟棄避免堆積一堆過期的任務(wù)把Worker壓垮。我一般會(huì)在任務(wù)配置里加上expires比如數(shù)據(jù)同步任務(wù)設(shè)置兩小時(shí)過期。這樣即使Worker因?yàn)槟撤N原因堵住了恢復(fù)之后也不會(huì)去執(zhí)行那些早就沒有意義的舊任務(wù)。3. 環(huán)境搭建從零到跑通第一個(gè)定時(shí)任務(wù)3.1 依賴安裝與Broker選型先把基礎(chǔ)依賴裝上。Celery本身很輕量核心包就一個(gè)pip install celery但光有Celery還不夠你需要一個(gè)Broker。常見的選擇是Redis和RabbitMQ我兩個(gè)都用過說說各自的適用場景。Redis的優(yōu)勢是部署簡單、性能好、大家普遍熟悉。如果你團(tuán)隊(duì)已經(jīng)在用Redis做緩存直接復(fù)用同一個(gè)實(shí)例用不同的db編號(hào)隔離是最省事的方案。缺點(diǎn)是Redis作為Broker時(shí)消息的可靠性保障不如RabbitMQ極端情況下比如Redis突然宕機(jī)且沒有持久化可能丟消息。RabbitMQ的優(yōu)勢是消息可靠性更強(qiáng)支持更復(fù)雜的路由策略適合對任務(wù)丟失零容忍的場景。缺點(diǎn)是運(yùn)維成本更高需要單獨(dú)維護(hù)一個(gè)RabbitMQ集群。我的建議是如果定時(shí)任務(wù)的結(jié)果不是“丟了就出大事”的那種用Redis就夠了如果是涉及資金、訂單等關(guān)鍵業(yè)務(wù)老老實(shí)實(shí)上RabbitMQ。我現(xiàn)在的項(xiàng)目用的是Redis因?yàn)槎〞r(shí)任務(wù)主要是數(shù)據(jù)同步和報(bào)表生成偶爾丟一次可以手動(dòng)補(bǔ)跑。安裝Redis的Python客戶端pip install redis3.2 Celery應(yīng)用的初始化配置接下來創(chuàng)建一個(gè)Celery應(yīng)用實(shí)例。我習(xí)慣把配置拆成獨(dú)立的文件方便不同環(huán)境開發(fā)、測試、生產(chǎn)用不同的參數(shù)。先建一個(gè)celery_app.pyfrom celery import Celery app Celery(myproject) app.config_from_object(celery_config) app.autodiscover_tasks([tasks])然后建一個(gè)celery_config.pybroker_url redis://:password127.0.0.1:6379/0 result_backend redis://:password127.0.0.1:6379/1 timezone Asia/Shanghai enable_utc False task_serializer json result_serializer json accept_content [json] task_track_started True task_time_limit 3600 task_soft_time_limit 3000 worker_max_tasks_per_child 200 worker_prefetch_multiplier 1這里有幾個(gè)配置值得展開說。timezone和enable_utc必須一起設(shè)置。Celery默認(rèn)用UTC時(shí)間如果你不改成東八區(qū)你寫的crontab(hour8)實(shí)際上會(huì)在北京時(shí)間下午四點(diǎn)執(zhí)行。這個(gè)坑我踩過當(dāng)時(shí)納悶了一整天為什么任務(wù)不在早上跑。task_track_started設(shè)為True之后任務(wù)開始執(zhí)行時(shí)狀態(tài)會(huì)變成STARTED方便你區(qū)分“排隊(duì)中”和“正在跑”。task_time_limit是硬超時(shí)到時(shí)間直接殺掉Worker進(jìn)程task_soft_time_limit是軟超時(shí)會(huì)拋出一個(gè)異常讓任務(wù)自己處理。我一般設(shè)軟超時(shí)比硬超時(shí)少幾分鐘給任務(wù)一個(gè)優(yōu)雅退出的機(jī)會(huì)。worker_prefetch_multiplier設(shè)為1很重要。默認(rèn)值是4意味著每個(gè)Worker會(huì)一次性從Broker預(yù)取4個(gè)任務(wù)。對于執(zhí)行時(shí)間較長的定時(shí)任務(wù)來說這會(huì)導(dǎo)致任務(wù)分配不均——一個(gè)Worker手里壓著好幾個(gè)任務(wù)慢慢做其他Worker閑著。設(shè)為1就是“做完一個(gè)再拿一個(gè)”。3.3 寫出第一個(gè)可用的定時(shí)任務(wù)配置好了之后寫一個(gè)最簡單的任務(wù)來驗(yàn)證鏈路是否通暢。在tasks/目錄下建一個(gè)demo.pyfrom celery_app import app import logging logger logging.getLogger(__name__) app.task(bindTrue, max_retries3, default_retry_delay60) def heartbeat(self): logger.info(心跳任務(wù)執(zhí)行中任務(wù)ID: %s, self.request.id) return ok然后在celery_config.py里注冊調(diào)度規(guī)則beat_schedule { heartbeat-every-minute: { task: tasks.demo.heartbeat, schedule: 60.0, }, }啟動(dòng)Beat和Worker開兩個(gè)終端# 終端1啟動(dòng)Beat celery -A celery_app beat -l info # 終端2啟動(dòng)Worker celery -A celery_app worker -l info -Q celery如果一切正常你會(huì)在Beat的日志里看到它每分鐘往Broker發(fā)一條任務(wù)消息在Worker的日志里看到任務(wù)被執(zhí)行并返回了結(jié)果。注意Beat進(jìn)程默認(rèn)會(huì)把調(diào)度狀態(tài)存到一個(gè)本地文件里celerybeat-schedule。如果你在多臺(tái)機(jī)器上同時(shí)啟動(dòng)Beat它們會(huì)各自維護(hù)自己的調(diào)度狀態(tài)導(dǎo)致任務(wù)重復(fù)觸發(fā)。Beat必須是單點(diǎn)的這一點(diǎn)后面還會(huì)展開講。4. 生產(chǎn)環(huán)境才會(huì)暴露的五個(gè)坑4.1 Beat單點(diǎn)問題與高可用方案前面提到Beat必須是單點(diǎn)但單點(diǎn)意味著如果Beat掛了所有定時(shí)任務(wù)都會(huì)停。這個(gè)問題在生產(chǎn)環(huán)境是致命的。我試過幾種方案。最簡單的是用supervisor守護(hù)Beat進(jìn)程掛了自動(dòng)重啟。但這只能解決進(jìn)程崩潰的情況如果服務(wù)器本身宕機(jī)還是會(huì)有空窗期。更可靠的方案是用分布式鎖保證同一時(shí)刻只有一個(gè)Beat在運(yùn)行。具體做法是啟動(dòng)多個(gè)Beat實(shí)例但每個(gè)實(shí)例在觸發(fā)任務(wù)之前先搶一把Redis鎖搶到鎖的才真正發(fā)送任務(wù)。Celery本身沒有內(nèi)置這個(gè)機(jī)制需要自己實(shí)現(xiàn)一個(gè)自定義的Scheduler類。from celery.beat import Scheduler import redis class LockedScheduler(Scheduler): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.redis redis.Redis(host127.0.0.1, port6379, db2) def tick(self): lock self.redis.lock(celery_beat_lock, timeout30) if lock.acquire(blockingFalse): try: return super().tick() finally: lock.release() return 0.0然后在啟動(dòng)Beat時(shí)指定這個(gè)Schedulercelery -A celery_app beat -S celery_app.LockedScheduler -l info這樣即使起了多個(gè)Beat同一時(shí)刻也只有一個(gè)在真正工作。不過說實(shí)話這個(gè)方案增加了復(fù)雜度如果你的定時(shí)任務(wù)不是特別關(guān)鍵用supervisor守護(hù)單點(diǎn)Beat 監(jiān)控告警也夠用了。4.2 任務(wù)重復(fù)執(zhí)行的冪等性處理定時(shí)任務(wù)最怕的就是重復(fù)執(zhí)行。網(wǎng)絡(luò)抖動(dòng)、Worker超時(shí)重啟、Beat重復(fù)觸發(fā)都可能導(dǎo)致同一個(gè)任務(wù)被執(zhí)行兩次。如果你的任務(wù)是“給用戶發(fā)優(yōu)惠券”重復(fù)執(zhí)行就是事故。解決思路是讓任務(wù)具備冪等性——執(zhí)行一次和執(zhí)行多次的結(jié)果是一樣的。具體做法有幾種第一種是用數(shù)據(jù)庫唯一約束。比如插入數(shù)據(jù)時(shí)用INSERT IGNORE或ON CONFLICT DO NOTHING重復(fù)執(zhí)行不會(huì)產(chǎn)生臟數(shù)據(jù)。第二種是在任務(wù)開頭檢查狀態(tài)。比如app.task def sync_order_data(date_str): if cache.get(fsync_done:{date_str}): logger.info(任務(wù)已執(zhí)行過跳過) return # 執(zhí)行同步邏輯 do_sync(date_str) cache.set(fsync_done:{date_str}, 1, ex86400)第三種是用Celery的task_id做去重。在Beat配置里給每個(gè)任務(wù)指定固定的task_idWorker端維護(hù)一個(gè)已執(zhí)行task_id的集合遇到重復(fù)的直接跳過。我一般用第二種簡單直接而且緩存過期時(shí)間設(shè)長一點(diǎn)比如24小時(shí)就能覆蓋絕大多數(shù)重復(fù)場景。4.3 Worker內(nèi)存泄漏與自動(dòng)重啟Python的垃圾回收機(jī)制在處理循環(huán)引用時(shí)有時(shí)候不夠及時(shí)長時(shí)間運(yùn)行的Worker進(jìn)程內(nèi)存會(huì)慢慢漲上去。我遇到過一次Worker跑了三天之后內(nèi)存占用從200MB漲到了2GB最后被系統(tǒng)OOM Killer干掉了。Celery提供了一個(gè)配置來解決這個(gè)問題worker_max_tasks_per_child 200這個(gè)配置的意思是Worker每執(zhí)行200個(gè)任務(wù)就自動(dòng)重啟一次子進(jìn)程釋放內(nèi)存。200這個(gè)數(shù)字可以根據(jù)你的任務(wù)內(nèi)存占用情況調(diào)整。如果單個(gè)任務(wù)內(nèi)存占用大就調(diào)小一點(diǎn)如果任務(wù)很輕量可以調(diào)大。還有一個(gè)相關(guān)的配置是worker_max_memory_per_child單位是KB。比如設(shè)為200000就是200MB子進(jìn)程內(nèi)存超過這個(gè)值就重啟。這個(gè)更精確但需要你對任務(wù)的內(nèi)存占用有大致了解。4.4 任務(wù)超時(shí)后的狀態(tài)不一致前面說了task_time_limit和task_soft_time_limit但這里有個(gè)隱蔽的問題硬超時(shí)殺掉Worker之后任務(wù)在Broker里的狀態(tài)可能還是“正在執(zhí)行”。如果你依賴Result Backend查任務(wù)狀態(tài)會(huì)看到任務(wù)永遠(yuǎn)卡在STARTED。解決辦法是在任務(wù)代碼里捕獲軟超時(shí)異常主動(dòng)更新狀態(tài)from celery.exceptions import SoftTimeLimitExceeded app.task(bindTrue, soft_time_limit3000, time_limit3600) def long_running_task(self): try: do_heavy_work() except SoftTimeLimitExceeded: self.update_state(stateFAILURE, meta{error: 任務(wù)超時(shí)}) raise這樣即使任務(wù)被強(qiáng)制終止?fàn)顟B(tài)也能被正確標(biāo)記。另外建議給Result Backend設(shè)置過期時(shí)間避免歷史任務(wù)狀態(tài)無限堆積result_expires 864004.5 時(shí)鐘漂移導(dǎo)致的任務(wù)偏移這個(gè)問題比較隱蔽。如果你的服務(wù)器時(shí)鐘和標(biāo)準(zhǔn)時(shí)間有偏差Beat觸發(fā)任務(wù)的時(shí)間就會(huì)不準(zhǔn)。我遇到過一臺(tái)服務(wù)器的時(shí)鐘慢了五分鐘導(dǎo)致所有定時(shí)任務(wù)都延遲了五分鐘執(zhí)行。解決辦法是配置NTP同步。在Linux上sudo apt install ntpdate sudo ntpdate pool.ntp.org或者用systemd的timesyncdsudo timedatectl set-ntp true另外Celery本身也有一個(gè)beat_max_loop_interval配置控制Beat檢查調(diào)度規(guī)則的最大間隔。默認(rèn)是300秒意味著Beat最多每5分鐘檢查一次有沒有任務(wù)要觸發(fā)。如果你的定時(shí)任務(wù)精度要求到分鐘級(jí)建議把這個(gè)值調(diào)小beat_max_loop_interval 605. 監(jiān)控與告警讓定時(shí)任務(wù)不再“黑盒”5.1 用Flower實(shí)時(shí)查看任務(wù)狀態(tài)Celery生態(tài)里最好用的監(jiān)控工具是Flower。它是一個(gè)Web界面能實(shí)時(shí)看到Worker狀態(tài)、任務(wù)執(zhí)行歷史、隊(duì)列長度等信息。安裝和啟動(dòng)pip install flower celery -A celery_app flower --port5555啟動(dòng)后訪問http://localhost:5555就能看到儀表盤。我一般會(huì)關(guān)注幾個(gè)指標(biāo)隊(duì)列積壓數(shù)量如果持續(xù)增長說明Worker不夠用、任務(wù)平均執(zhí)行時(shí)間突然變長可能是數(shù)據(jù)量增大或代碼有問題、失敗任務(wù)列表及時(shí)發(fā)現(xiàn)問題。Flower還支持配置認(rèn)證生產(chǎn)環(huán)境記得加上celery -A celery_app flower --basic_authadmin:password5.2 任務(wù)失敗后的告警接入光有監(jiān)控界面還不夠沒人會(huì)一直盯著屏幕看。需要配置告警任務(wù)失敗時(shí)主動(dòng)通知。Celery提供了任務(wù)失敗的回調(diào)鉤子可以在配置里指定def on_task_failure(self, exc, task_id, args, kwargs, einfo): send_alert(f任務(wù) {self.name} 執(zhí)行失敗任務(wù)ID: {task_id}錯(cuò)誤: {exc}) app.conf.task_annotations { *: {on_failure: on_task_failure} }send_alert可以對接郵件、企業(yè)微信機(jī)器人、釘釘機(jī)器人等。我一般用企業(yè)微信機(jī)器人簡單可靠import requests def send_alert(message): webhook https://qyapi.weixin.qq.com/cgi-bin/webhook/send?keyyour_key requests.post(webhook, json{ msgtype: text, text: {content: message} })注意告警要設(shè)置頻率限制避免任務(wù)大面積失敗時(shí)被告警轟炸。可以在send_alert里加一個(gè)簡單的計(jì)數(shù)器同一任務(wù)短時(shí)間內(nèi)失敗多次只發(fā)一條告警。5.3 任務(wù)執(zhí)行日志的集中管理Celery的日志默認(rèn)輸出到stdout在生產(chǎn)環(huán)境需要收集起來方便排查。我一般用兩種方式結(jié)合一是把Worker日志寫到文件用logrotate做切割celery -A celery_app worker -l info --logfile/var/log/celery/worker.log二是任務(wù)內(nèi)部的關(guān)鍵日志直接寫到數(shù)據(jù)庫或日志服務(wù)方便按任務(wù)ID檢索。比如app.task(bindTrue) def process_data(self, date_str): logger.info(開始處理 %s 的數(shù)據(jù)任務(wù)ID: %s, date_str, self.request.id) try: result do_process(date_str) logger.info(處理完成結(jié)果: %s, result) except Exception as e: logger.error(處理失敗: %s, e, exc_infoTrue) raise這樣出問題的時(shí)候拿著任務(wù)ID就能在日志系統(tǒng)里搜到完整的執(zhí)行鏈路。6. 性能調(diào)優(yōu)與擴(kuò)展思路6.1 Worker并發(fā)數(shù)的合理設(shè)置Worker的并發(fā)數(shù)由--concurrency參數(shù)控制默認(rèn)是CPU核心數(shù)。但這個(gè)默認(rèn)值不一定適合你的場景。如果任務(wù)是CPU密集型的比如圖像處理、大量計(jì)算并發(fā)數(shù)設(shè)為CPU核心數(shù)或核心數(shù)1比較合適再多會(huì)因?yàn)樯舷挛那袚Q反而變慢。如果任務(wù)是IO密集型的比如網(wǎng)絡(luò)請求、數(shù)據(jù)庫讀寫并發(fā)數(shù)可以設(shè)大很多因?yàn)閃orker大部分時(shí)間在等IO。我一般設(shè)成CPU核心數(shù)的2到4倍。celery -A celery_app worker -l info --concurrency8另外如果你的任務(wù)對內(nèi)存敏感可以用--poolprefork默認(rèn)配合worker_max_tasks_per_child來控制。如果任務(wù)非常輕量且數(shù)量巨大可以考慮--poolgevent或--pooleventlet用協(xié)程提高并發(fā)。6.2 任務(wù)路由與隊(duì)列隔離當(dāng)定時(shí)任務(wù)種類變多之后不同類型的任務(wù)混在同一個(gè)隊(duì)列里會(huì)互相影響。比如一個(gè)耗時(shí)的報(bào)表生成任務(wù)堵住了隊(duì)列后面所有輕量的心跳檢測任務(wù)都跟著延遲。解決辦法是用隊(duì)列隔離。在配置里定義多個(gè)隊(duì)列task_routes { tasks.report.*: {queue: report}, tasks.sync.*: {queue: sync}, tasks.heartbeat: {queue: default}, }然后啟動(dòng)不同的Worker監(jiān)聽不同的隊(duì)列celery -A celery_app worker -Q report -c 2 -l info celery -A celery_app worker -Q sync -c 4 -l info celery -A celery_app worker -Q default -c 2 -l info這樣報(bào)表任務(wù)再慢也不會(huì)影響心跳檢測。代價(jià)是需要維護(hù)更多的Worker進(jìn)程根據(jù)實(shí)際情況權(quán)衡。6.3 大規(guī)模任務(wù)下的Beat性能當(dāng)beat_schedule里的規(guī)則超過幾百條時(shí)Beat的調(diào)度性能會(huì)開始下降。因?yàn)锽eat是單線程的每次tick都要遍歷所有規(guī)則判斷是否該觸發(fā)。優(yōu)化思路有幾個(gè)。一是合并同類任務(wù)比如把“每分鐘檢查一次A”和“每分鐘檢查一次B”合并成一個(gè)任務(wù)內(nèi)部依次執(zhí)行。二是減少不必要的規(guī)則定期清理已經(jīng)廢棄的定時(shí)任務(wù)。三是用自定義Scheduler做分片把規(guī)則分散到多個(gè)Beat實(shí)例上每個(gè)實(shí)例只負(fù)責(zé)一部分規(guī)則。我目前的項(xiàng)目有大概80條調(diào)度規(guī)則Beat運(yùn)行很穩(wěn)定CPU占用可以忽略不計(jì)。所以除非你的規(guī)則數(shù)量真的很大否則不用太擔(dān)心這個(gè)問題。7. 一些零散但重要的經(jīng)驗(yàn)關(guān)于任務(wù)參數(shù)的設(shè)計(jì)我建議盡量傳簡單的、可序列化的參數(shù)。Celery默認(rèn)用JSON序列化傳不了復(fù)雜的Python對象。如果需要傳復(fù)雜數(shù)據(jù)傳一個(gè)ID讓任務(wù)自己去數(shù)據(jù)庫查。關(guān)于時(shí)區(qū)再強(qiáng)調(diào)一次timezone和enable_utc一定要配對設(shè)置。我見過太多人只設(shè)了timezone忘了enable_utc結(jié)果任務(wù)執(zhí)行時(shí)間怎么都對不上。關(guān)于Beat的調(diào)度文件celerybeat-schedule這個(gè)文件不要隨便刪。它記錄了上次調(diào)度的時(shí)間戳刪掉之后Beat會(huì)重新計(jì)算可能導(dǎo)致任務(wù)重復(fù)觸發(fā)。如果確實(shí)需要重置先停掉Beat再刪。關(guān)于任務(wù)命名建議用完整的模塊路徑比如tasks.report.generate_daily不要用簡寫。這樣在Flower里排查問題時(shí)能一眼看出任務(wù)在哪個(gè)文件里。關(guān)于測試Celery任務(wù)可以用task_always_eager配置在本地同步執(zhí)行方便調(diào)試task_always_eager True # 僅開發(fā)環(huán)境使用這樣調(diào)用task.delay()會(huì)直接在當(dāng)前進(jìn)程執(zhí)行不走Broker斷點(diǎn)調(diào)試很方便。但記得生產(chǎn)環(huán)境一定要關(guān)掉。最后說一個(gè)我踩過的坑不要在任務(wù)函數(shù)里直接操作全局變量或數(shù)據(jù)庫連接。Celery的Worker是多進(jìn)程模型每個(gè)子進(jìn)程有獨(dú)立的內(nèi)存空間全局變量在子進(jìn)程之間不共享。數(shù)據(jù)庫連接也要用連接池或者在任務(wù)內(nèi)部創(chuàng)建和關(guān)閉避免連接泄漏。這套Celery定時(shí)任務(wù)體系我已經(jīng)跑了快一年管理著六十多個(gè)定時(shí)任務(wù)日均執(zhí)行幾千次整體很穩(wěn)定。中間遇到過的問題基本都在這篇文章里了希望對正在搭建類似系統(tǒng)的你有幫助。如果只能記住一件事那就是Beat單點(diǎn)、任務(wù)冪等、失敗告警這三點(diǎn)做好了定時(shí)任務(wù)就不會(huì)出大問題。