邊看這篇 Celery – Best Practices 邊做的簡單摘錄與筆記。
- 不要用資料庫當作 AMQP Broker。Celery 會建立數個 process 去 poll 資料庫來檢查是否有新的工作，這會導致資料庫的 disk I/O 增加，也會增加對資料庫的連接數目。
- 使用更多佇列 (不要只用一個)。
並不是所有的 task 執行時間、次數跟權重都一樣，例如不重要的 task A 可能會執行很多次，但比較重要的 task B 只有零星幾個。一個佇列會導致 Celery 依序執行佇列裡的工作，所以前面可能會排了很多 task A 工作，就多花了許多時間執行，反而重要的 task B 工作延後了。依照 task 分佇列擺放，可以避免這樣的情況。 - 使用權重。Celery 可以針對佇列設定權重與分流，可以使用專門的 worker 來服務指定的佇列，讓 task 得到更好的服務。
CELERY_QUEUES = ( Queue('default', Exchange('default'), routing_key='default'), Queue('for_task_A', Exchange('for_task_A'), routing_key='for_task_A'), Queue('for_task_B', Exchange('for_task_B'), routing_key='for_task_B'), ) CELERY_ROUTES = { 'my_taskA': {'queue': 'for_task_A', 'routing_key': 'for_task_A'}, 'my_taskB': {'queue': 'for_task_B', 'routing_key': 'for_task_B'}, }celery worker -E -l INFO -n workerA -Q for_task_A celery worker -E -l INFO -n workerB -Q for_task_B - 使用 Celery 的錯誤處理機制。task 可以指定這些參數 default_retry_delay=300, max_retries=5 來指定重試間隔與重試次數。task 裏面只要使用 try…except 跟 self.retry 就可以了。
@app.task(bind=True, default_retry_delay=300, max_retries=5) def my_task_A(): try: print("doing stuff here...") except SomeNetworkException as e: print("maybe do some clenup here....") self.retry(e) # Retry! - 使用 Flower。這是一個只要裝上就能使用的 Module，可以用來觀察 Task/Queue 的狀況。
- 只有在真的需要時，才保留執行結果。不需要的話，就加上 CELERY_IGNORE_RESULT = True，Celery 會自動丟棄結果。
- 不要傳遞 ORM 物件給 task。這是因為 Celery 是用 serialization 方式來傳遞參數到別的 Process (Task 是在其他的 Process 上執行)，預設可以使用 pickle, cPickle, JSON, YAML ，但是 serialization/deserialization 是有負擔的，而且不保證所有狀態都能保存，建議最好是 pure 的物件或是用整數、字串等比較不容易出狀況的型態。