FastAPI-backend-kehityksen intensiivikurssi · Oppitunti

Ajastetut ja toistuvat työt Celery Beatilla

Suorittakaa toistuvia töitä Celery Beatin avulla ja koordinoikaa cron-tyyppiset aikataulut turvallisesti useiden työntekijöiden välillä.

Oppitunti 4/413 vaihetta

Ajastetut ja toistuvat työt Celery Beatilla on ilmainen FastAPI-backend-kehityksen intensiivikurssi-oppitunti CoddyKitissä. Tämä on oppitunti 4/4. Voit lukea tästä oppimispolusta kokonaan mitkä tahansa 3 oppituntia ilmaiseksi — sen jälkeen CoddyKit PRO avaa kaikki oppitunnit sekä käytännön harjoittelun sisäänrakennetulla koodieditorilla ja ympäri vuorokauden toimivalla tekoälytuutorilla. Oppitunti kuuluu FastAPI-backend-kehityksen intensiivikurssi-oppimispolkuun, ja edistymisesi synkronoituu verkon ja CoddyKit-sovelluksen välillä. FastAPI-backend-kehityksen intensiivikurssi-kurssilla on yhteensä 4 oppituntia.

Miksi Celery Beat?

FastAPI-sovelluksesi käynnistää kertaluonteisia taustatehtäviä Celeryllä ongelmitta, mutta joidenkin tehtävien on suorituttava ajastetusti: lähetä koontisähköposti joka aamu, vanhenna hylätyt ostoskorit 10 minuutin välein tai laske analytiikka uudelleen joka yö.

Celery Beat on ajastusprosessi. Se ei suorita tehtäviä itse — se herää jokaisella kellopulssilla, päättää, mitkä tehtävät ovat erääntyneet, ja lähettää ne välittäjälle (Redis/RabbitMQ). Tavalliset celery worker -prosessit poimivat ne ja suorittavat ne.

Beat = kello, joka julkaisee erääntyneet tehtävät.
Worker = voima, joka suorittaa ne.

Tämä erottelu on koko syy siihen, että ajastetut työt skaalautuvat: yksi Beat ja monta työntekijää.

Ajastuksen määrittäminen

Ajastukset sijaitsevat Celery-sovelluksessa kohdassa conf.beat_schedule. Kukin merkintä yhdistää nimen sanakirjaan, joka sisältää task-polun, schedule-asetuksen (sekunteina, timedelta-arvona tai crontabina) sekä valinnaiset args/kwargs-arvot.

Yksinkertaisin ajastus on kiinteä aikaväli. Alla puhdistustehtävä suoritetaan 30 sekunnin välein. Tehtävän polkumerkkijonon on vastattava täsmälleen työntekijän rekisteröityä nimeä.

from celery import Celery
from datetime import timedelta

app = Celery("jobs", broker="redis://localhost:6379/0")

@app.task(name="tasks.cleanup_sessions")
def cleanup_sessions():
    # delete expired sessions from the DB
    return "cleaned"

app.conf.beat_schedule = {
    "cleanup-every-30s": {
        "task": "tasks.cleanup_sessions",
        "schedule": timedelta(seconds=30),
    },
}

Cron-tyyliset ajastukset crontab()-funktiolla

Kiinteät aikavälit ovat karkeita. Käytä varsinaiseen kalenterilogiikkaan — esimerkiksi "joka arkipäivä kello 07:30" — celery.schedules.crontab-funktiota. Se vastaa Unix cronin kenttiä: minute, hour, day_of_week, day_of_month, month_of_year.

crontab(minute=0, hour=0) — keskiyöllä joka päivä.
crontab(minute="*/15") — 15 minuutin välein.
crontab(hour=7, minute=30, day_of_week="1-5") — 07:30 maanantaista perjantaihin.

Asettamatta jätettyjen kenttien oletusarvo on * (kaikki arvot), aivan kuten crontab-rivillä.

from celery.schedules import crontab

app.conf.beat_schedule = {
    "morning-digest": {
        "task": "tasks.send_digest",
        "schedule": crontab(hour=7, minute=30, day_of_week="1-5"),
    },
    "quarter-hour-sync": {
        "task": "tasks.sync_inventory",
        "schedule": crontab(minute="*/15"),
    },
}

Argumenttien ja asetusten välittäminen merkintäkohtaisesti

Kukin ajastusmerkintä voi sisältää argumentteja ja kutsukohtaisia asetuksia. Parametrisoi sama tehtävä eri tavoin eri merkinnöissä käyttämällä args- (sijaintiin perustuvat) tai kwargs-argumentteja (nimettyihin parametreihin perustuvat). options-sanakirjalla voit reitittää ajastetun tehtävän tiettyyn queue-jonoon, asettaa expires-arvon tai ohittaa prioriteetin.

expires on tärkeä ajastetuissa tehtävissä: jos Beat asettaa tehtävän jonoon mutta työntekijöillä on ruuhkaa, vanhentunut viesti hylätään sen sijaan, että se suoritettaisiin myöhässä ja tehtäviä kasautuisi.

from celery.schedules import crontab

app.conf.beat_schedule = {
    "warm-cache-eu": {
        "task": "tasks.warm_cache",
        "schedule": crontab(minute="*/5"),
        "kwargs": {"region": "eu-west"},
        "options": {"queue": "cache", "expires": 120},
    },
    "warm-cache-us": {
        "task": "tasks.warm_cache",
        "schedule": crontab(minute="*/5"),
        "kwargs": {"region": "us-east"},
        "options": {"queue": "cache", "expires": 120},
    },
}

Ajastusten rekisteröinti dekoraattorilla

Yhden suuren beat_schedule-sanakirjan sijaan voit rekisteröidä merkinnät tehtävien lähellä käyttämällä on_after_configure-signaalia ja app.add_periodic_task-metodia. Näin ajastus pysyy sen käynnistämän koodin yhteydessä ja vanhentuneita merkkijonopolkuja syntyy vähemmän.

add_periodic_task(schedule, signature, name=...) saa ensin aikavälin tai crontabin ja sen jälkeen tehtävän allekirjoituksen. Voit välittää task.s(arg)-kutsun argumenttien esiasettamista varten.

from celery import Celery
from celery.schedules import crontab

app = Celery("jobs", broker="redis://localhost:6379/0")

@app.task
def rotate_logs(target):
    return f"rotated {target}"

@app.on_after_configure.connect
def setup_periodic_tasks(sender, **kwargs):
    sender.add_periodic_task(
        crontab(hour=3, minute=0),
        rotate_logs.s("app.log"),
        name="nightly-log-rotation",
    )

Aikavyöhykkeet: ajastusten yleisin virhe

Oletuksena Celery tulkitsee crontab-ajat UTC-aikana. Jos kirjoitat crontab(hour=7) odottaen paikallista aikaa 07:00, tehtävä käynnistyy väärään kellonaikaan. Aseta aikavyöhyke aina nimenomaisesti ja tee valinta harkiten.

timezone — vyöhyke, jossa crontab-ajat arvioidaan.
enable_utc=True — pidä sisäiset aikaleimat UTC-ajassa (suositus), mutta arvioi crontabit edelleen valitsemassasi timezone-aikavyöhykkeessä.

Kiinnitä aikavyöhyke määrityksiin, jotta kaikki kehittäjät ja palvelimet käyttävät samaa vyöhykettä isäntäkoneen paikallisesta aikavyöhykkeestä riippumatta.

app.conf.update(
    timezone="Europe/Istanbul",
    enable_utc=True,
)

# crontab(hour=9, minute=0) now means 09:00 Europe/Istanbul,
# stored/transmitted internally as UTC.

Beatin suorittaminen ja pysyvä ajastustiedosto

Käynnistät ajastimen omana prosessinaan. Oletusajastin tallentaa viimeisimmän suorituksen tilan pieneen tiedostoon, jotta se ei käynnistä kaikkea uudelleen palvelimen uudelleenkäynnistyksen jälkeen.

celery -A app beat -l info — suorita Beat.
--schedule /var/run/celerybeat-schedule — määrittää shelve-tilatiedoston sijainnin.
Kehitysympäristössä voit suorittaa työntekijän ja Beatin yhdessä komennolla celery -A app worker -B, mutta älä koskaan käytä -B-asetusta tuotannossa — se sitoo kellon yhden työntekijän elinkaareen.

Tuotannossa suorita täsmälleen yksi Beat-prosessi. Kaksi Beatia tarkoittaa, että jokainen ajastettu tehtävä suoritetaan kahdesti.

Yhden ajastimen sääntö työntekijöiden kesken

Voit skaalata työntekijöitä vaakasuunnassa kymmeniin podeihin asti, ja Celery huolehtii lopusta: välittäjä jakaa jokaisen jonossa olevan viestin täsmälleen yhdelle työntekijälle. Mutta Beat on kello, ja käytössä saa olla vain yksi kello.

Jos kaksi Beat-prosessia on käynnissä, kumpikin päättää itsenäisesti tehtävän olevan erääntynyt ja julkaisee sen, joten tilaajat näkevät kaksoissuorituksia. Yleisiä tapoja, joilla tämä tapahtuu vahingossa:

Kaksi podin replikaa käynnistää kumpikin beat-prosessin.
Käytät asetusta worker -B ja skaalaat kyseisen työntekijädeploymentin sen jälkeen vähintään kahteen replikaa.

Korjaus: erillinen Beat Deployment, jossa on replicas: 1, erillään työntekijä-Deploymentista, jota voit skaalata vapaasti.

Idempotenssi: suunnittele tehtävät kestämään kaksoissuoritukset

Kaksoissuorituksia tapahtuu yhden Beatin käytöstä huolimatta — Beat voi käynnistyä uudelleen väärällä hetkellä, välittäjä voi toimittaa viestin uudelleen vähintään kerran -periaatteella tai operaattori voi tehdä virheen. Vankka suoja on tehdä ajastetuista tehtävistä idempotentteja: kahdesti suorittaminen tuottaa saman vaikutuksen kuin kerran suorittaminen.

Yksinkertainen malli on lyhytkestoinen hajautettu lukko Redisissä käyttämällä SET NX-komentoa ja vanhenemisaikaa. Lukon saanut suorittaa työn; samanaikaiset tai päällekkäiset suoritukset ohitetaan turvallisesti.

import redis

r = redis.Redis(host="localhost", port=6379, db=0)

@app.task(name="tasks.charge_subscriptions")
def charge_subscriptions():
    # acquire a lock valid for 300s; only one runner proceeds
    got = r.set("lock:charge_subscriptions", "1", nx=True, ex=300)
    if not got:
        return "skipped: already running"
    try:
        # ... perform the billing run exactly once ...
        return "charged"
    finally:
        r.delete("lock:charge_subscriptions")

Tietokantapohjaiset ajastukset: django-celery-beat / RedBeat

Oletustiedostoajastin tarkoittaa, että ajastuksen muuttaminen edellyttää koodin muokkaamista ja Beatin uudelleenkäynnistämistä. Dynaamisia ajastuksia varten tarvitset taustalla toimivan tallennuspaikan, jota voit muokata ajon aikana.

RedBeat — tallentaa ajastukset Redisiin; erinomainen vaihtoehto Redis-pohjaiseen FastAPI-pinoon. Aseta beat_scheduler = "redbeat.RedBeatScheduler".
django-celery-beat — tallentaa merkinnät SQL-tauluihin, joita voi muokata hallintokäyttöliittymässä.

RedBeat tarjoaa myös Redis-pohjaisen lukon, joten jos käynnistät vahingossa kaksi Beatia, vain toinen toimii aktiivisena ajastimena — tämä on turvaverkko yhden kellon säännölle.

app.conf.update(
    redbeat_redis_url="redis://localhost:6379/1",
    beat_scheduler="redbeat.RedBeatScheduler",
    beat_max_loop_interval=5,
)

Itsenäinen tarkistus cron-tehtävän erääntymisestä

Jotta saisit käsityksen siitä, mitä Beat tekee jokaisella kellopulssilla, tässä on pieni itsenäinen simulaatio: annettuna luettelo cron-tyyppisistä töistä (aikaväli sekunteina ja viimeisimmän suorituksen aikaleima), päätä, mitkä niistä ovat juuri nyt erääntyneet. Tämä on Beatin suorittaman ydinsilmukan periaate — vertaa aikaa "seuraavaan erääntymiseen" — ilman välittäjää.

def due_jobs(now, jobs):
    fired = []
    for name, interval, last_run in jobs:
        if now - last_run >= interval:
            fired.append(name)
    return fired

jobs = [
    ("cleanup", 30, 0),
    ("digest", 3600, 3500),
    ("sync", 900, 100),
]

now = 1000
print(due_jobs(now, jobs))  # ['cleanup', 'sync']

Pikatarkistus: päällekkäisten ajastettujen suoritusten estäminen

Otat FastAPI- ja Celery-pinon käyttöön Kubernetesissa. Skaalaat työntekijä-Deploymentin kuuteen neljään replikan sijaan kuormituksen käsittelemiseksi, ja jokainen työntekijä käynnistetään komennolla celery -A app worker -B. Ajastetut tehtävät alkavat suorittua neljä kertaa kukin. Mikä on oikea korjaus?

Vastaa interaktiivisessa oppitunnissa

Kertaus

Voit nyt ajastaa toistuvia tehtäviä Celery Beatin avulla:

Beat ajoittaa, workerit suorittavat — Beat julkaisee erääntyneet tehtävät välittäjälle; yksi Beat, monta workeria.
Ajastukset määritetään kohteessa beat_schedule tai funktion add_periodic_task avulla; käytä aikaväleihin timedelta-objektia ja kalenterilogiikkaan crontab()-funktiota.
Kohtaiset args/options-asetukset parametrisoivat tehtäviä, ja queue/expires-asetusten avulla vältetään vanhentuneiden tehtävien kasaantuminen.
Aikavyöhykkeet ovat oletusarvoisesti UTC — määritä timezone ja enable_utc eksplisiittisesti.
Vain yksi kello — älä koskaan käytä tuotannossa komentoa worker -B; suorita yksi erillinen Beat-replika.
Idempotenttius (Redis-tekniikan SET NX -lukot) suojaa päällekkäisiltä tai uudelleen toimitetuilta suorituksilta.
RedBeat / django-celery-beat tarjoavat ajonaikaisesti muokattavat, pysyvät ja lukoilla suojatut ajastukset.

Aloita maksutta

Opi FastAPI-backend-kehityksen intensiivikurssi tekoälytuutorin avulla — ilmaiseksi

Kirjoita ja suorita oikeaa koodia selaimessa, saa välitöntä apua tekoälytuutorilta ympäri vuorokauden ja jatka siitä, mihin jäit, verkossa tai sovelluksessa.

Kurssit: 21
Oppitunnit: 84

Aloita opiskelu maksutta Näytä koko kurssi

Usein kysytyt kysymykset

Onko oppitunti ”Ajastetut ja toistuvat työt Celery Beatilla” ilmainen?

Kyllä — voit lukea täällä verkossa kokonaan ilmaiseksi mitkä tahansa FastAPI-backend-kehityksen intensiivikurssi-oppimispolun 3 oppituntia, myös oppitunnin “Ajastetut ja toistuvat työt Celery Beatilla”. Sen jälkeen CoddyKit PRO avaa kaikki oppitunnit sekä interaktiiviset harjoitukset sisäänrakennetulla koodieditorilla ja ympäri vuorokauden toimivalla tekoälytuutorilla. FastAPI-backend-kehityksen intensiivikurssi-kurssilla on yhteensä 4 oppituntia.

Mitä opin oppitunnilla ”Ajastetut ja toistuvat työt Celery Beatilla”?

Suorittakaa toistuvia töitä Celery Beatin avulla ja koordinoikaa cron-tyyppiset aikataulut turvallisesti useiden työntekijöiden välillä. Harjoittelet FastAPI-backend-kehityksen intensiivikurssi-aihetta koodilla, jonka suoritat suoraan selaimessa. Ympäri vuorokauden käytettävissä oleva tekoälytuutori vastaa kysymyksiisi oppitunnin aikana.

Tarvitsenko kokemusta aloittaakseni FastAPI-backend-kehityksen intensiivikurssi-opiskelun?

Aiempi kokemus ei ole tarpeen. CoddyKitin FastAPI-backend-kehityksen intensiivikurssi-oppimispolku sopii vasta-alkajista edistyneisiin, joten voit aloittaa tästä tai alusta ja edetä omaan tahtiisi. Tämä on oppitunti 4/4.

Kuinka kauan ”Ajastetut ja toistuvat työt Celery Beatilla”-oppitunnin suorittaminen kestää?

Useimmat CoddyKitin oppitunnit kestävät noin 5–10 minuuttia. Jokainen oppitunti on lyhyt ja interaktiivinen, joten edistyt tasaisesti ja voit jatkaa siitä, mihin jäit – sekä verkossa että sovelluksessa.

Voinko kirjoittaa ja suorittaa koodia tällä FastAPI-backend-kehityksen intensiivikurssi-oppitunnilla?

Kyllä. Jokainen FastAPI-backend-kehityksen intensiivikurssi-oppitunti sisältää sisäänrakennetun koodieditorin, joten voit kirjoittaa ja suorittaa oikeaa koodia suoraan selaimessa ja saada välitöntä palautetta tekoälyltä – paikallista asennusta ei tarvita.

Kaikki tämän kurssin oppitunnit

Kevyt työn siirtäminen BackgroundTasksilla
Celery-työntekijöiden kytkeminen FastAPI-sovellukseen
Uudelleenyritykset, idempotenssi ja hylättyjen viestien käsittely
Ajastetut ja toistuvat työt Celery Beatilla

← Takaisin: FastAPI-backend-kehityksen intensiivikurssi