因?yàn)轫?xiàng)目是后期引入celery,所以導(dǎo)致構(gòu)建docker的時(shí)候只有fastapi的項(xiàng)目，celery的重啟比較麻煩
1.docker安裝celery

pip install celery

安裝celery的時(shí)候注意python版本與celery版本的適配，有些celery的版本不支持python的版本，具體的版本請(qǐng)看celery官網(wǎng)里面的版本信息

2.在工程目錄中創(chuàng)建celery的啟動(dòng)文件，這里我創(chuàng)建的是tasks.py 文件

from celery import Celerycelery_app = Celery("worker",broker="redis://:frasergen2022@192.168.2.189:26379/0",backend="redis://:frasergen2022@192.168.2.189:26379/0",include=["apps.tools.my_celery.__init__"]
)

這是celery的啟動(dòng)文件，里面最好不要引入你的項(xiàng)目里面的變量，如果引入你項(xiàng)目里的變了，后期去定義后臺(tái)任務(wù)的時(shí)候，引入celery_app會(huì)陷入到循環(huán)引用的深坑

3.測(cè)試celery

celery -A tasks.celery_app worker --loglevel=info

4.可以后臺(tái)啟動(dòng)celery

celery multi start w1 -A tasks.celery_app -l info --logfile=celerylog.log

這里不用啟動(dòng)，后續(xù)會(huì)用看門(mén)狗(watchdog)監(jiān)控任務(wù)文件，如果文件修改，會(huì)重啟celery

5.安裝看門(mén)狗(watchdog)

pip install watchdog

參考鏈接：https://whoosy.cn/2019/08/01/Celery/celery使用/
后臺(tái)啟動(dòng)watchdog

nohup watchmedo auto-restart --directory=/data/cloud_platform/apps/tools/my_celery/ --pattern=*.py --recursive -- celery -A tasks.celery_app worker --loglevel=info --logfile=celerylog.log > watchmedo.log 2> watchmedo.elog &

–directory : 監(jiān)控路徑
–pattern： 監(jiān)控文件后綴

6.配置celery后臺(tái)任務(wù)
1.配置tortoise-orm數(shù)據(jù)庫(kù)連接

import asynciofrom tortoise import Tortoise
from celery.signals import worker_process_init, worker_process_shutdownfrom apps.models import User
from tasks import celery_app
from apps.tools.db_config import ORM_LINK_CONFasync def init_db():await Tortoise.init(config=ORM_LINK_CONF)@worker_process_init.connect
def on_worker_init(*args, **kwargs):print('初始化數(shù)據(jù)庫(kù)')from celery._state import _task_stackif _task_stack.top is not None:loop = _task_stack.top.request.loopelse:loop = asyncio.get_event_loop()loop.run_until_complete(init_db())@worker_process_shutdown.connect
def on_worker_shutdown(*args, **kwargs):print('關(guān)閉數(shù)據(jù)庫(kù)')from celery._state import _task_stackif _task_stack.top is not None:loop = _task_stack.top.request.loopelse:loop = asyncio.get_event_loop()loop.run_until_complete(Tortoise.close_connections())

2.配置后臺(tái)任務(wù)

@celery_app.task(name="get_user_task")
def get_user_task(*args, **kwargs):asyncio.get_event_loop().run_until_complete(_get_user_task())async def _get_user_task():user = await User.filter().all()for item in user:print(item.nickname, item.username)print(f"{item.phone=}")

代碼上傳到docker后，celery會(huì)重啟

7.測(cè)試接口+后臺(tái)異步任務(wù)

@router.get("/get_user_info", summary="測(cè)試后臺(tái)任務(wù)獲取用戶信息")
async def get_user_info():get_user_task.delay()return res()

結(jié)果：