Thinking more...: Celery Best Practice 筆記

邊看這篇 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 的物件或是用整數、字串等比較不容易出狀況的型態。

Thinking more...

星期一, 4月 18, 2016

Celery Best Practice 筆記

沒有留言: