Django?celery的替代品Dramatiq使用特性詳解

更新時(shí)間：2023年06月21日 11:04:35 作者：alue

這篇文章主要為大家介紹了Django?celery的替代品Dramatiq使用特性詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪

Django 框架消息隊(duì)列 celery

Django 框架需要一個(gè)消息隊(duì)列來(lái)支持分布式任務(wù)處理，大部分人會(huì)選擇 celery。但 celery 不支持 windows。

之前，用過(guò)funboost，功能強(qiáng)大，但是其代碼充斥著拼寫(xiě)錯(cuò)誤，讓人難以忍受。

今天看了 Dramatiq，發(fā)現(xiàn)完全滿(mǎn)足項(xiàng)目需求。于是上手嘗試了一下。

celery 特性

如果用了普羅米修斯中間件，則可以通過(guò) http://127.0.0.1:9191/ ，查看workers狀態(tài)。（在windows上試了，響應(yīng)不穩(wěn)定）
代碼變動(dòng)后，可向主進(jìn)程發(fā)送 SIGHUP 信號(hào)，從而讓 workers 更新?tīng)顟B(tài)。（但windows下，沒(méi)有有向進(jìn)程發(fā)送信號(hào)的便捷手段）

$ kill -s HUP 13047

可以加入 --watch 參數(shù)，實(shí)現(xiàn) hot-reload, 但生產(chǎn)環(huán)境不推薦。
max_retries：設(shè)定最大重試次數(shù)。
retry_when：設(shè)定重試條件。
max_age ：任務(wù)壽命毫秒。

當(dāng)消息超過(guò)最大重試次數(shù)或者在隊(duì)列里時(shí)長(zhǎng)超過(guò)壽命，則會(huì)進(jìn)入死信區(qū)，在這里最多保留 7 天，然后被丟棄。

time_limit：執(zhí)行時(shí)長(zhǎng)上限（毫秒），默認(rèn)為 10 分鐘。超時(shí)，則會(huì)報(bào)出 TimeLimitExceeded 異常。
priority : 任務(wù)優(yōu)先級(jí)，默認(rèn)為 0，越小優(yōu)先級(jí)越高。
send_with_options()：提供任務(wù)調(diào)度能力
方便與Django框架結(jié)合，在admin中能夠看到任務(wù)執(zhí)行歷史。

Actors

Actor 的并行

你的 actor 會(huì)跟其它 actors 并發(fā)執(zhí)行，注意操作相同資源時(shí)，可能出現(xiàn)的競(jìng)爭(zhēng)關(guān)系。

簡(jiǎn)單消息參數(shù)

給 actor 傳遞的消息應(yīng)該是能被 json 序列化的數(shù)據(jù)，例如 bool, int, float, bytes, string, list and dict.

冪等性

受網(wǎng)絡(luò)、硬件、電源等因素影響，一個(gè) actor 可能會(huì)反復(fù)接收到相同消息，所以你要保證多次執(zhí)行 actor 的安全性。

Messages 消息

brokers 接收到消息后，會(huì)持久化到硬盤(pán)。broker 重啟后，會(huì)重新激活消息。

消息處理結(jié)果，可以通過(guò) Results 中間件，返回 Redis 或者 Memcached 中。

優(yōu)雅的處理中斷

import dramatiq
from dramatiq.middleware import Interrupt
@dramatiq.actor(max_retries=3, notify_shutdown=True)
def long_running_task():
    try:
        setup()
        do_work()
    except Shutdown:
        cleanup()
        raise