Python FastAPI+Celery+RabbitMQ實現(xiàn)分布式圖片水印處理系統(tǒng)
更新時間：2025年04月04日 09:23:59 作者：老大白菜
這篇文章主要為大家詳細介紹了Python FastAPI如何結合Celery以及RabbitMQ實現(xiàn)簡單的分布式圖片水印處理系統(tǒng),感興趣的小伙伴可以跟隨小編一起學習一下
實現(xiàn)思路
- FastAPI 服務器
- Celery 任務隊列
- RabbitMQ 作為消息代理
- 定時任務處理
完整步驟
首先創(chuàng)建項目結構：
c:\Users\Administrator\Desktop\meitu\
├── app/
│ ├── __init__.py
│ ├── main.py
│ ├── celery_app.py
│ ├── tasks.py
│ └── config.py
├── requirements.txt
└── celery_worker.py
1.首先創(chuàng)建 requirements.txt：
fastapi==0.104.1 uvicorn==0.24.0 celery==5.3.4 python-dotenv==1.0.0 requests==2.31.0
2.創(chuàng)建配置文件：
from dotenv import load_dotenv
import os
load_dotenv()
# RabbitMQ配置
RABBITMQ_HOST = os.getenv("RABBITMQ_HOST", "localhost")
RABBITMQ_PORT = os.getenv("RABBITMQ_PORT", "5672")
RABBITMQ_USER = os.getenv("RABBITMQ_USER", "guest")
RABBITMQ_PASS = os.getenv("RABBITMQ_PASS", "guest")
# Celery配置
CELERY_BROKER_URL = f"amqp://{RABBITMQ_USER}:{RABBITMQ_PASS}@{RABBITMQ_HOST}:{RABBITMQ_PORT}//"
CELERY_RESULT_BACKEND = "rpc://"
# 定時任務配置
CELERY_BEAT_SCHEDULE = {
'process-images-every-hour': {
'task': 'app.tasks.process_images',
'schedule': 3600.0, # 每小時執(zhí)行一次
},
'daily-cleanup': {
'task': 'app.tasks.cleanup_old_images',
'schedule': 86400.0, # 每天執(zhí)行一次
}
}
3.創(chuàng)建 Celery 應用：
from celery import Celery
from app.config import CELERY_BROKER_URL, CELERY_RESULT_BACKEND, CELERY_BEAT_SCHEDULE
celery_app = Celery(
'image_processing',
broker=CELERY_BROKER_URL,
backend=CELERY_RESULT_BACKEND,
include=['app.tasks']
)
# 配置定時任務
celery_app.conf.beat_schedule = CELERY_BEAT_SCHEDULE
celery_app.conf.timezone = 'Asia/Shanghai'
4.創(chuàng)建任務文件：
from app.celery_app import celery_app
from app.watermark import ImageWatermarker
import os
from datetime import datetime, timedelta
@celery_app.task
def add_watermark_task(image_path, text, position='center'):
"""異步添加水印任務"""
watermarker = ImageWatermarker()
try:
result_path = watermarker.add_watermark(
image_path=image_path,
text=text,
position=position
)
return {"status": "success", "output_path": result_path}
except Exception as e:
return {"status": "error", "message": str(e)}
@celery_app.task
def process_images():
"""定時處理圖片任務"""
image_dir = "images/pending"
if not os.path.exists(image_dir):
return {"status": "error", "message": "Pending directory not found"}
processed = 0
for image in os.listdir(image_dir):
if image.lower().endswith(('.png', '.jpg', '.jpeg')):
add_watermark_task.delay(
os.path.join(image_dir, image),
"自動處理水印",
'center'
)
processed += 1
return {"status": "success", "processed": processed}
@celery_app.task
def cleanup_old_images():
"""清理舊圖片任務"""
output_dir = "images/processed"
if not os.path.exists(output_dir):
return {"status": "error", "message": "Output directory not found"}
threshold_date = datetime.now() - timedelta(days=7)
cleaned = 0
for image in os.listdir(output_dir):
image_path = os.path.join(output_dir, image)
if os.path.getctime(image_path) < threshold_date.timestamp():
os.remove(image_path)
cleaned += 1
return {"status": "success", "cleaned": cleaned}
5.創(chuàng)建 FastAPI 應用：
from fastapi import FastAPI, File, UploadFile, BackgroundTasks
from fastapi.responses import JSONResponse
import os
from app.tasks import add_watermark_task
from app.celery_app import celery_app
app = FastAPI(title="圖片水印處理服務")
@app.post("/upload/")
async def upload_image(
file: UploadFile = File(...),
text: str = "水印文本",
position: str = "center"
):
# 保存上傳的文件
file_path = f"images/uploads/{file.filename}"
os.makedirs(os.path.dirname(file_path), exist_ok=True)
with open(file_path, "wb") as buffer:
content = await file.read()
buffer.write(content)
# 創(chuàng)建異步任務
task = add_watermark_task.delay(file_path, text, position)
return JSONResponse({
"status": "success",
"message": "圖片已上傳并加入處理隊列",
"task_id": task.id
})
@app.get("/task/{task_id}")
async def get_task_status(task_id: str):
task = celery_app.AsyncResult(task_id)
if task.ready():
return {"status": "completed", "result": task.result}
return {"status": "processing"}
@app.get("/tasks/scheduled")
async def get_scheduled_tasks():
return {"tasks": celery_app.conf.beat_schedule}
6.創(chuàng)建 Celery worker 啟動文件：
from app.celery_app import celery_app
if __name__ == '__main__':
celery_app.start()
使用說明
首先安裝依賴：
pip install -r requirements.txt
確保 RabbitMQ 服務已啟動
啟動 FastAPI 服務器：
uvicorn app.main:app --reload
啟動 Celery worker：
celery -A celery_worker.celery_app worker --loglevel=info
啟動 Celery beat（定時任務）：
celery -A celery_worker.celery_app beat --loglevel=info
這個系統(tǒng)提供以下功能：
- 通過 FastAPI 接口上傳圖片并異步處理水印
- 使用 Celery 處理異步任務隊列
- 使用 RabbitMQ 作為消息代理
- 支持定時任務：
- 每小時自動處理待處理圖片
- 每天清理一周前的舊圖片
- 支持任務狀態(tài)查詢
- 支持查看計劃任務列表
API 端點：
- POST /upload/ - 上傳圖片并創(chuàng)建水印任務
- GET /task/{task_id} - 查詢?nèi)蝿諣顟B(tài)
- GET /tasks/scheduled - 查看計劃任務列表
以上就是Python FastAPI+Celery+RabbitMQ實現(xiàn)分布式圖片水印處理系統(tǒng)的詳細內(nèi)容，更多關于Python圖片水印的資料請關注腳本之家其它相關文章！
相關文章
今天小編就為大家分享一篇基于pytorch 預訓練的詞向量用法詳解，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2020-01-01
這篇文章主要為大家介紹了Python中的Dict，具有一定的參考價值，感興趣的小伙伴們可以參考一下，希望能夠給你帶來幫助2021-12-12
這篇文章主要給大家分享的是100 個 Python 小例子，前期已經(jīng)給大家分過100個小例子的（一）和（二），今天小編繼續(xù)和大家分享（三），希望歲正在學習的你有所幫助2022-01-01
python實現(xiàn)微秒級等待問題(windows)
這篇文章主要介紹了python實現(xiàn)微秒級等待問題(windows),具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2024-06-06
人工智能這么火，0基礎能學python嗎？python該怎么選擇編輯器？怎么搭建python運行環(huán)境？python好學嗎，怎么學？這是所有python入門前同學都會提出的疑問，這篇文章和大家一起學習python，感興趣的小伙伴們可以加入2018-03-03
這篇文章主要介紹了Python腳本實現(xiàn)蝦米網(wǎng)簽到功能的方法,涉及Python調(diào)用URL模塊實現(xiàn)數(shù)據(jù)傳輸與處理的相關技巧,需要的朋友可以參考下2016-04-04
這篇文章主要為大家詳細介紹了如何利用Python中的PyQt5自制一個PDF轉(zhuǎn)PNG格式圖片的小工具，文中的示例代碼講解詳細，感興趣的可以了解一下2023-02-02
這篇文章主要介紹了快速了解Python相對導入，具有一定借鑒價值,需要的朋友可以參考下2018-01-01
Python使用Matplotlib模塊時坐標軸標題中文及各種特殊符號顯示方法
這篇文章主要介紹了Python使用Matplotlib模塊時坐標軸標題中文及各種特殊符號顯示方法,結合具體實例分析了Python使用Matplotlib模塊過程中針對中文及特殊符號的顯示方法,需要的朋友可以參考下2018-05-05
今天小編就為大家分享一篇Python 移動光標位置的方法，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2019-01-01