溫馨提示×

Flask中怎么實(shí)現(xiàn)消息隊(duì)列

小億
147
2024-05-10 18:16:57
欄目: 編程語言

在Flask中實(shí)現(xiàn)消息隊(duì)列通常使用Celery這樣的異步任務(wù)隊(duì)列庫。以下是一個(gè)基本的示例：

  1. 首先，安裝Celery庫：
pip install celery
  1. 創(chuàng)建一個(gè)Flask應(yīng)用并配置Celery：
from flask import Flask
from celery import Celery

app = Flask(__name__)

# 配置Celery
app.config['CELERY_BROKER_URL'] = 'redis://localhost:6379/0'
app.config['CELERY_RESULT_BACKEND'] = 'redis://localhost:6379/0'

celery = Celery(app.name, broker=app.config['CELERY_BROKER_URL'])
celery.conf.update(app.config)
  1. 創(chuàng)建一個(gè)Celery任務(wù)：
@celery.task
def send_email(to, subject, body):
    # 發(fā)送郵件的邏輯
    pass
  1. 在Flask應(yīng)用中調(diào)用Celery任務(wù)：
from tasks import send_email

@app.route('/send_email')
def send_email_view():
    send_email.delay('example@example.com', 'Subject', 'Body')
    return 'Email sent!'
  1. 啟動(dòng)Celery worker：
celery worker -A app.celery --loglevel=info

這樣就可以在Flask應(yīng)用中使用Celery實(shí)現(xiàn)消息隊(duì)列功能了。當(dāng)調(diào)用send_email.delay時(shí)，任務(wù)會被添加到Celery隊(duì)列中，Celery worker會異步執(zhí)行這個(gè)任務(wù)。

0 踩