Taskok
A Client és az Enqueue metódusa önmagában elég egy nyers *asynq.Task beküldéséhez, de minden réteg, ami a queue-ra épül, az események modul listener-fan-outjától az ütemező cron-bejegyzéseiig, még három dolgot megoszt: egy közös enqueue-opció-készletet, egy közös boríték-formátumot az event-payloadokhoz, és egy közös task-típus-elnevezési sémát. Ez az oldal mindhármat tárgyalja; a mindennapokban változatlanul ezekkel találkozol mindenhol, ahol a queue alattuk van.
Enqueue-opciók
A queue az asynq opciók hasznos részhalmazát exportálja újra, hogy a gyakori esetekhez ne kelljen asynq-ot importálni:
| Opció | Hatás |
|---|---|
OnQueue(name) | nevesített queue-ba irányít (a worker WORKER_QUEUES súlyai szerint); üresen default |
MaxRetry(n) | ennyi újrapróbálkozás után kerül archívba a hibázó task |
Timeout(d) | egyetlen futási kísérlet időkorlátja |
Deadline(t) | abszolút határidő az összes kísérletre |
ProcessIn(d) | a feldolgozás d-vel későbbre tolása |
Unique(ttl) | azonos típus+payload task elnyomása, amíg az előző függőben van |
Retention(d) | a kész task még d-ig megmarad a Valkeyben, visszanézhetően |
TaskID(id) | explicit task-azonosító; az ütközés ErrDuplicate (idempotens kézbesítés alapja) |
Az opciók mindenhol ugyanígy néznek ki, ahol task születik: az Enqueue-nál, az events.Listen listener-regisztrációnál és az ütemező bejegyzéseinél. A shop mail queue-ja például a listener regisztrációjában jelenik meg:
events.Listen(reg, "sendOrderConfirmation", sendOrderConfirmation(deps), queue.OnQueue("mail"))
Ha a teljes asynq opciókészlet kell, a queue.AsynqOptions(opts...) ezeket nyers []asynq.Option-né alakítja, és az Enqueue mellett az asynq API-ja is nyitva áll (escape hatch, nem tiltott zóna).
A boríték (Envelope)
Minden event-task payloadja ugyanaz a wire-formátum, a queue.Envelope:
type Envelope struct {
ID string `json:"id"` // dispatch-azonosító, egyben asynq TaskID
Name string `json:"name"` // az event neve (EventName())
Payload json.RawMessage `json:"payload"` // a JSON-kódolt event-érték
Metadata map[string]string `json:"metadata,omitempty"` // W3C trace-kontextus
OccurredAt time.Time `json:"occurred_at"`
}
A boríték a dispatchertől (vagy az outboxtól) a fan-out taskon át minden listener-taskig változatlanul utazik: a listener ugyanazt az ID-t, payloadot és trace-kontextust látja, amit a producer beletett. Az env.Task() rendereli fan-out taskká, a queue.DecodeEnvelope(task) bontja ki a túloldalon.
Trace-propagáció a queue-n át
Az env.InjectTrace(ctx) a context trace-kontextusát W3C formátumban (traceparent/tracestate) a Metadata-ba írja; a worker oldalán az env.ExtractTrace(ctx) folytatja ugyanazt a trace-t. Így a HTTP-kérés → outbox → relay → worker → listener út egyetlen trace, a shopban a rendelésfeladástól a visszaigazoló e-mailig. A dispatcherek és a worker middleware ezt automatikusan megteszik; a részletekhez lásd az OpenTelemetry oldalt.
Task-típusok
Négy task-típus-prefix fedi le mindazt, ami a queue-n át folyik az async kategóriában, és az alábbi helperek az egyetlen forrásai a wire-neveknek:
| Helper | Task-típus | Ki hozza létre |
|---|---|---|
EventTaskType(name) | event:shop.orderPlaced | dispatch / outbox relay (a fan-out kiindulópontja) |
ListenerTaskType(event, listener) | listener:shop.orderPlaced:sendOrderConfirmation | a worker fan-out handlere, listenerenként egy |
ScheduleTaskType(name) | schedule:shop.someEvent | az ütemező, ütemezett event tüzelésekor |
JobTaskType(name) | job:shop.nightlySalesReport | az ütemező, ütemezett job tüzelésekor |
A kit workere a prefixek alapján osztja szét a taskokat a négy beépített handlerének; az IsEventTaskType/IsListenerTaskType/IsJobTaskType/IsScheduleTaskType és a JobName helperek ugyanezt a besorolást adják bárkinek, aki (pl. WithMux-szal regisztrált saját handlerben) nyers taskokkal dolgozik. Ezek a nevek jelennek meg az Asynqmonban és a worker logjaiban is: ránézésre látszik, melyik task melyik event melyik listenere. Az események modul és a worker chassis az, ami ténylegesen létrehozza és feldolgozza a taskokat ezekkel a nevekkel; ez az oldal csak az elnevezést definiálja, magát a fan-outot nem.
Mikor nyúlsz hozzá közvetlenül
- Saját, nem event-alapú task: építs
asynq.NewTask(...)-ot, add be aclient.Enqueue-tel, és regisztrálj rá handlert a workerbenWithMux-szal. - Idempotens „pontosan egyszer ható" enqueue:
TaskID+ErrDuplicate-kezelés, ahogy az Áttekintés oldalon. - Minden más (event dispatch, ütemezés) a felsőbb rétegek dolga: kezdd az eseményeknél.