Kem Intensif Pembangunan Bahagian Belakang FastAPI · Pelajaran

Menyambungkan Pekerja Celery kepada Aplikasi FastAPI

Konfigurasikan Celery dengan broker Redis, takrifkan tugasan dan hantar kerja daripada titik akhir dengan bahagian belakang hasil.

Pelajaran 2 daripada 413 langkah

Menyambungkan Pekerja Celery kepada Aplikasi FastAPI ialah pelajaran Kem Intensif Pembangunan Bahagian Belakang FastAPI percuma di CoddyKit. Ini ialah pelajaran 2 daripada 4. Sebanyak 3 pelajaran dalam laluan pembelajaran ini boleh dibaca sepenuhnya secara percuma — selepas itu, CoddyKit PRO membuka akses kepada semua pelajaran, serta latihan praktikal dengan penyunting kod terbina dalam dan tutor kecerdasan buatan yang tersedia 24/7. Pelajaran ini merupakan sebahagian daripada laluan pembelajaran Kem Intensif Pembangunan Bahagian Belakang FastAPI, dan kemajuan anda disegerakkan merentas web serta aplikasi CoddyKit. Kursus Kem Intensif Pembangunan Bahagian Belakang FastAPI merangkumi sejumlah 4 pelajaran.

Mengapa Celery dalam Aplikasi FastAPI?

Pengendali permintaan FastAPI mesti memberikan respons dengan cepat. Kerja seperti menghantar e-mel, menjana laporan, mengubah saiz imej atau memanggil API pihak ketiga yang lambat boleh mengambil masa beberapa saat, lalu menyekat proses worker dan menjejaskan daya pemprosesan.

Celery ialah baris gilir tugasan teragih. Anda meletakkan kerja ke dalam baris gilir, kemudian kumpulan proses worker yang berasingan menjalankannya di luar kitaran permintaan/respons.

Perantara — pengangkutan mesej yang menyimpan kerja beratur (kami menggunakan Redis).
Worker — proses yang berjalan lama dan mengambil serta melaksanakan tugasan.
Bahagian belakang hasil — stor pilihan untuk nilai pulangan dan keadaan tugasan.

Titik akhir kekal pantas dan hanya menjawab: "diterima, ini id kerja anda."

Memasang Komponen

Pasang Celery dengan tambahan Redis, serta pelayan Redis yang boleh dicapai oleh API dan worker anda.

celery[redis] memasang Celery dan klien redis.
Jalankan Redis secara setempat dengan Docker: docker run -p 6379:6379 redis.

Kedua-dua proses FastAPI dan proses worker mengimport objek aplikasi Celery yang sama, jadi kedua-duanya mesti berkongsi pangkalan kod dan URL perantara yang sama.

# requirements.txt
fastapi
uvicorn[standard]
celery[redis]
redis

# install
# pip install -r requirements.txt
# run redis: docker run -p 6379:6379 redis:7

Mencipta Aplikasi Celery

Takrifkan satu tika aplikasi Celery dalam modulnya sendiri (biasanya worker.py atau celery_app.py). Ia memerlukan nama, URL perantara dan URL bahagian belakang hasil.

broker — tempat tugasan dimasukkan ke dalam baris gilir (pangkalan data Redis 0).
backend — tempat hasil/keadaan disimpan (pangkalan data Redis 1, diasingkan untuk kejelasan).

Argumen pertama ("worker") menjadi awalan lalai untuk nama tugasan.

# celery_app.py
from celery import Celery

celery_app = Celery(
    "worker",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/1",
)

celery_app.conf.update(
    task_serializer="json",
    result_serializer="json",
    accept_content=["json"],
    timezone="UTC",
    enable_utc=True,
)

Mentakrifkan Tugasan Pertama Anda

Tugasan hanyalah fungsi yang dihiasi dengan @celery_app.task. Apabila dipanggil secara biasa, fungsi itu berjalan sebaris; apabila dipanggil dengan .delay() atau .apply_async(), fungsi itu disiri dan dihantar ke perantara.

Argumen mesti boleh disiri sebagai JSON (gunakan id dan jenis primitif, bukan objek ORM).
Berikan tugasan itu name yang jelas supaya penamaan semula fungsi kemudian tidak merosakkan mesej yang sedang beratur.

# tasks.py
import time
from celery_app import celery_app

@celery_app.task(name="tasks.send_report")
def send_report(user_id: int, email: str) -> dict:
    # simulate slow work
    time.sleep(5)
    return {"user_id": user_id, "sent_to": email, "status": "done"}

Menghantar Kerja dari Titik Akhir

Dalam laluan FastAPI, panggil .delay(...) untuk memasukkan tugasan ke dalam baris gilir. Ini mengembalikan AsyncResult serta-merta — ia tidak menunggu kerja selesai.

Balas dengan task.id dan HTTP 202 Accepted, yang menandakan bahawa permintaan telah diterima untuk diproses tetapi masih belum selesai.

Jangan sekali-kali memanggil .get() untuk mendapatkan hasil dalam pengendali permintaan — tindakan itu menyekat gelung peristiwa sehingga kerja selesai dan menggagalkan tujuan keseluruhan.

# main.py
from fastapi import FastAPI
from pydantic import BaseModel, EmailStr
from tasks import send_report

app = FastAPI()

class ReportRequest(BaseModel):
    user_id: int
    email: EmailStr

@app.post("/reports", status_code=202)
def create_report(req: ReportRequest):
    task = send_report.delay(req.user_id, req.email)
    return {"task_id": task.id, "status": "queued"}

Menjalankan Worker

Pelayan FastAPI hanya menghasilkan mesej. Anda mesti memulakan proses worker yang berasingan untuk menggunakan dan melaksanakannya.

celery_app selepas -A merujuk kepada modul dan tika.
--loglevel=info memaparkan setiap tugasan apabila diterima dan berjaya.
--concurrency=4 mengawal bilangan tugasan yang berjalan secara selari.

Untuk isu percabangan pada Windows atau macOS, tambahkan --pool=solo semasa pembangunan.

# terminal 1: API
uvicorn main:app --reload

# terminal 2: worker
celery -A celery_app.celery_app worker --loglevel=info --concurrency=4

Meninjau Keadaan Tugasan dengan Bahagian Belakang Hasil

Oleh sebab kita telah mengkonfigurasi bahagian belakang hasil, kita boleh mencari keadaan dan nilai pulangan kerja kemudian menggunakan idnya.

Bina AsyncResult daripada id tersebut dan ikatkannya pada aplikasi Celery yang sama. Keadaan yang berguna:

PENDING — tidak diketahui/belum bermula.
STARTED — telah diambil oleh worker.
SUCCESS — selesai; .result mengandungi nilai pulangan.
FAILURE — pengecualian telah dicetuskan.

Klien meninjau titik akhir keadaan ini sehingga tugasan sedia.

# main.py (continued)
from celery.result import AsyncResult
from celery_app import celery_app

@app.get("/reports/{task_id}")
def get_status(task_id: str):
    result = AsyncResult(task_id, app=celery_app)
    payload = {"task_id": task_id, "state": result.state}
    if result.successful():
        payload["result"] = result.result
    return payload

apply_async: Kiraan Undur, ETA dan Cubaan Semula

.delay(*args) ialah bentuk ringkas bagi .apply_async(args=...). Bentuk yang lebih panjang membuka pilihan penjadualan dan penghalaan:

countdown=10 — tunggu 10 saat sebelum melaksanakan.
eta=datetime(...) — jalankan pada masa tertentu.
queue="emails" — halakan ke baris gilir bernama.
retry=True dengan retry_policy — cuba semula apabila berlaku ralat perantara.

from tasks import send_report

send_report.apply_async(
    args=[42, "user@example.com"],
    countdown=10,
    queue="reports",
)

Mencuba Semula Tugasan yang Gagal

Kegagalan sementara (API yang tidak stabil, tamat masa) patut dicuba semula, bukan dibiarkan hilang. Ikat tugasan dengan bind=True supaya self tersedia, kemudian panggil self.retry().

max_retries mengehadkan bilangan cubaan.
default_retry_delay atau countdown memberikan tempoh jeda antara cubaan.
autoretry_for boleh mencuba semula secara automatik untuk pengecualian tertentu.

Selepas semua cubaan semula habis, tugasan berakhir dalam keadaan FAILURE.

from celery_app import celery_app

@celery_app.task(
    bind=True,
    name="tasks.charge_card",
    max_retries=3,
    autoretry_for=(ConnectionError,),
    retry_backoff=True,
)
def charge_card(self, order_id: int):
    try:
        process_payment(order_id)
    except ConnectionError as exc:
        raise self.retry(exc=exc, countdown=5)

FastAPI BackgroundTasks berbanding Celery

FastAPI menyediakan pembantu BackgroundTasks yang ringan. Ia menjalankan kerja dalam proses yang sama, selepas respons dihantar. Ketahui kesesuaian setiap satu:

BackgroundTasks — kerja pantas yang dilancarkan dan dilupakan (menghantar satu e-mel, menulis log). Tiada cubaan semula, tiada penjejakan hasil dan kerja hilang jika proses dimulakan semula.
Celery — kerja berat, lama, boleh dicuba semula atau boleh dijadualkan yang memerlukan kekekalan, penskalaan mendatar merentas mesin serta keterlihatan hasil/keadaan.

Peraturan mudah: jika kehilangan kerja akibat ranap tidak boleh diterima, atau kerja itu berat dari segi CPU/masa, gunakan Celery.

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

def write_log(message: str):
    with open("audit.log", "a") as f:
        f.write(message + "\n")

@app.post("/click")
def click(bt: BackgroundTasks):
    bt.add_task(write_log, "user clicked")
    return {"ok": True}

Simulasi Baris Gilir Kendiri

Anda tidak boleh menjalankan perantara sebenar dalam penilai dalam talian, tetapi corak penghasil/pengguna di sebalik Celery adalah mudah. Contoh Python tulen ini meniru idea tersebut: kerja dimasukkan ke dalam baris gilir, worker mengosongkan baris gilir, dan hasil dikumpulkan mengikut id — tepat seperti aliran yang diautomatikkan oleh Celery melalui Redis.

from collections import deque

queue = deque()
results = {}

def enqueue(task_id, user_id, email):
    queue.append((task_id, user_id, email))
    results[task_id] = "PENDING"

def worker():
    while queue:
        task_id, user_id, email = queue.popleft()
        results[task_id] = {
            "user_id": user_id,
            "sent_to": email,
            "status": "SUCCESS",
        }

enqueue("t1", 42, "a@x.com")
enqueue("t2", 7, "b@x.com")
worker()

for tid in ("t1", "t2"):
    print(tid, results[tid])

Semakan Ringkas

Uji pemahaman anda tentang cara titik akhir FastAPI menyerahkan kerja kepada Celery.

Jawab dalam pelajaran interaktif

Rumusan

Anda telah menyambungkan Celery ke dalam aplikasi FastAPI dari awal hingga akhir:

Mencipta satu tika Celery dengan perantara Redis dan bahagian belakang hasil.
Mentakrifkan tugasan yang boleh disiri sebagai JSON dengan @celery_app.task serta nama yang jelas.
Menghantar kerja dari titik akhir dengan .delay(), lalu mengembalikan id tugasan dan 202 Accepted tanpa menyekat.
Menjalankan proses celery ... worker yang berasingan untuk menggunakan baris gilir.
Meninjau keadaan dan hasil kerja melalui AsyncResult.
Menggunakan apply_async untuk kiraan undur/baris gilir dan self.retry() untuk cubaan semula yang tahan lasak.
Memilih antara BackgroundTasks FastAPI (ringan, dalam proses) dan Celery (kekal, boleh diskalakan, boleh dicuba semula).

Peraturan emas: titik akhir memasukkan kerja ke dalam baris gilir dan memberikan respons dengan cepat; worker melakukan kerja berat.

Percuma untuk bermula

Pelajari Kem Intensif Pembangunan Bahagian Belakang FastAPI dengan tutor kecerdasan buatan — percuma

Tulis dan jalankan kod sebenar dalam pelayar anda, dapatkan bantuan segera daripada tutor kecerdasan buatan yang tersedia 24/7, dan sambung semula dari tempat anda berhenti di web atau dalam aplikasi.

Kursus: 21
Pelajaran: 84

Mula belajar secara percuma Lihat kursus penuh

Soalan Lazim

Adakah pelajaran “Menyambungkan Pekerja Celery kepada Aplikasi FastAPI” percuma?

Ya — sebanyak 3 pelajaran dalam laluan pembelajaran Kem Intensif Pembangunan Bahagian Belakang FastAPI, termasuk “Menyambungkan Pekerja Celery kepada Aplikasi FastAPI”, boleh dibaca sepenuhnya secara percuma di web ini. Selepas itu, CoddyKit PRO membuka akses kepada semua pelajaran, serta latihan interaktif dengan penyunting kod terbina dalam dan tutor kecerdasan buatan yang tersedia 24/7. Kursus Kem Intensif Pembangunan Bahagian Belakang FastAPI merangkumi sejumlah 4 pelajaran.

Apakah yang akan saya pelajari dalam “Menyambungkan Pekerja Celery kepada Aplikasi FastAPI”?

Konfigurasikan Celery dengan broker Redis, takrifkan tugasan dan hantar kerja daripada titik akhir dengan bahagian belakang hasil. Anda berlatih Kem Intensif Pembangunan Bahagian Belakang FastAPI menggunakan kod praktikal yang dijalankan terus dalam pelayar, manakala tutor kecerdasan buatan 24/7 menjawab soalan anda semasa anda mengikuti pelajaran.

Adakah saya memerlukan pengalaman untuk memulakan Kem Intensif Pembangunan Bahagian Belakang FastAPI?

Tiada pengalaman terdahulu diperlukan. Pembelajaran Kem Intensif Pembangunan Bahagian Belakang FastAPI di CoddyKit disusun untuk pelajar daripada peringkat pemula hingga lanjutan, jadi anda boleh bermula di sini atau dari awal dan belajar mengikut kadar anda sendiri. Ini ialah pelajaran 2 daripada 4.

Berapa lamakah pelajaran “Menyambungkan Pekerja Celery kepada Aplikasi FastAPI” diambil?

Kebanyakan pelajaran CoddyKit mengambil masa kira-kira 5–10 minit. Setiap pelajaran ringkas dan interaktif, jadi anda boleh membuat kemajuan secara berterusan dan menyambung tepat dari tempat anda berhenti di web atau aplikasi.

Bolehkah saya menulis dan menjalankan kod dalam pelajaran Kem Intensif Pembangunan Bahagian Belakang FastAPI ini?

Ya. Setiap pelajaran Kem Intensif Pembangunan Bahagian Belakang FastAPI menyertakan penyunting kod terbina dalam, jadi anda boleh menulis dan menjalankan kod sebenar terus dalam pelayar serta menerima maklum balas kecerdasan buatan serta-merta — tanpa memerlukan persediaan setempat.

Semua pelajaran dalam kursus ini

Pengalihan Kerja Ringan dengan BackgroundTasks
Menyambungkan Pekerja Celery kepada Aplikasi FastAPI
Percubaan Semula, Keidempotenan dan Pengendalian Surat Mati
Kerja Berjadual dan Berkala dengan Celery Beat

← Kembali ke Kem Intensif Pembangunan Bahagian Belakang FastAPI