fastapi 後台任務定時任務與訊息佇列

總結有時候我們不可避免的要建立一些非同步任務，這些任務與主線程無關或比較耗時不必立即反饋結果或不需要反饋結果。

from fastapi import backgroundtasks, fastapi
)def
write_notification
(email:
str, message="")
:# 提前定義好任務
with
open
("log.txt"
, mode=
"w")
as email_file:
content = f"notification for : "
email_file.write(content)
"/send-notification/"
)async
defsend_notification
(email:
str, background_tasks: backgroundtasks)
:# 請求乙個後台任務的方法，
background_tasks.add_task(write_notification, email, message=
"some notification"
)return

後台任務對應的是比較簡單的一些情況。

這裡我們使用到的是rearq，這個包是根據arq進行的二次開發，在api上面參考了celery，以做到更人性化和簡單化。參考頁面

在fastapi中初始化rearq

配置settings

# settings.py #在settings資料夾裡面配置rearq需要的引數，rearq使用redis rearq =

初始化

# 在專案的根目錄__init__初始化rearq
from rearq import rearq
from demo import settings# 上文的settings.py
rearq = rearq(
**settings.rearq,
)

啟動startup和shutdown

# factory.py 
from demo import rearq, settings
#在startup裡面啟動需要使用的相關元件(在定時任務裡面使用了redis快取和tortoise-orm非同步orm，所以要提前啟動)
@rearq.on_startup
async
defon_startup()
:await asyncredisutil.init(
**settings.redis)
# redis非同步操作的元件
await tortoise.init(config=settings.tortoise_orm)
# tortoise-orm非同步操作的元件
@rearq.on_shutdown
async
defon_shutdown()
:await asyncredisutil.close(
)await tortoise.close_connections(
)async
def():
...

建立定時任務

#tasks.py
from enum import enum
class
queue
(str
, enum)
:# 使用到的佇列
task =
"task"
withdraw =
"withdraw"
notify =
"notify"
pass_task =
"pass_task"
@rearq.task(queue=queue.withdraw.value, cron=
"*/5 * * * *"
)#每五分鐘執行一次，使用的佇列是withdraw
async
deftiming_withdraw
(self)
:"""
定時任務,self必須填，其他的為該方法需要的引數
:param self:
:return:
"""..
.return..
.#會寫入日誌...吧？
@rearq.task(queue=queue.task.value,
)async
defcheck_receive_task
( self, uaid, task_id, uid, platform_category: db.taskplatformcategory, category: db.taskcategory):
# 除了self，其他的引數都是在使用這個佇列任務的時候需要填充的..
.return
"未找到"

使用任務

在建立完任務之後，使用任務的方式如下：

#router.py
#介面@router.get(
"/test_queue"
)async
deftest_queue()
: check_receive_task.delay(..
.)#這裡填寫需要的引數，這樣就會放到訊息佇列裡面等待執行
return

引入日誌

rearq.log.init_logging(verbose)
[source]

啟動rearq

專案一般使用的是docker-compose進行管理，一般的專案結構大概如下

version :'3' services : demo: #主服務 build : . env_file :- .env ports:- '8008:8000' depends_on :- redis image : demo command -port 8000 - -host 0.0.0.0 timing-worker: #啟動定時任務服務 env_file :- .env image : demo depends_on :- demo command : rearq worker demo.factory:rearq - -timer task-worker: #啟動任務服務，每個queue單獨啟動 env_file :- .env image : demo depends_on :- demo command : rearq worker demo.factory:rearq -q task withdraw-worker: #啟動任務服務，每個queue單獨啟動 env_file :- .env image : demo depends_on :- demo command : rearq worker demo.factory:rearq -q withdraw ...redis: #啟動redis hostname : redis image : redis volumes

:- ./redis/data:/data

非docker-compose的啟動方式：

command: rearq worker demo.factory:rearq --timer

注意：定時任務要單獨啟動乙個worker，每個queue也要啟動乙個worker,timer的工作是到時間把任務推到對應的queue裡面去執行，所以必須單獨啟動乙個。

更具體的任務使用方式檢視參考頁面

fastapi 後台任務定時任務與訊息佇列

NC65後台任務（定時任務）

Linux學習之後台任務與定時任務二十

linux後台定時任務

fastapi 後台任務 定時任務與訊息佇列

NC65後台任務（定時任務）

Linux學習之後台任務與定時任務 二十

linux後台定時任務

相關推薦

fastapi 後台任務定時任務與訊息佇列

Linux學習之後台任務與定時任務二十