Kolejki#

Vapor Queues (vapor/queues) to napisany w czystym Swifcie system kolejkowania, który pozwala przenieść odpowiedzialność za zadania na osobnego workera.

Niektóre z zadań, do których ten pakiet dobrze się nadaje:

• Wysyłanie e-maili poza głównym wątkiem requestu

• Wykonywanie złożonych lub długotrwałych operacji na bazie danych

• Zapewnianie integralności i odporności zadań

• Przyspieszanie czasu odpowiedzi poprzez opóźnianie przetwarzania niekrytycznego

• Planowanie zadań do wykonania o określonym czasie

Ten pakiet jest podobny do Ruby Sidekiq. Zapewnia następujące funkcje:

• Bezpieczną obsługę sygnałów SIGTERM i SIGINT wysyłanych przez dostawców hostingu w celu wskazania zamknięcia, restartu lub nowego wdrożenia.

• Różne priorytety kolejek. Na przykład możesz określić, że dane zadanie kolejki ma być uruchamiane na kolejce email, a inne zadanie na kolejce przetwarzania danych.

• Implementuje niezawodny proces kolejkowania, który pomaga radzić sobie z nieoczekiwanymi awariami.

• Zawiera funkcję maxRetryCount, która powtórzy zadanie aż do jego powodzenia, do określonej liczby prób.

• Wykorzystuje NIO, aby wykorzystać wszystkie dostępne rdzenie i EventLoopy do zadań.

• Pozwala użytkownikom planować zadania cykliczne

Queues posiada obecnie jeden oficjalnie wspierany driver, który komunikuje się z głównym protokołem:

• QueuesRedisDriver

Queues posiada również drivery tworzone przez społeczność:

• QueuesMongoDriver

• QueuesFluentDriver

Pierwsze kroki#

Zobaczmy, jak zacząć korzystać z Queues.

Pakiet#

Pierwszym krokiem do korzystania z Queues jest dodanie jednego z driverów jako zależności do Twojego projektu w pliku manifestu SwiftPM. W tym przykładzie użyjemy drivera Redis.

// swift-tools-version:5.8
import PackageDescription

let package = Package(
    name: "MyApp",
    dependencies: [
        /// Any other dependencies ...
        .package(url: "https://github.com/vapor/queues-redis-driver.git", from: "1.0.0"),
    ],
    targets: [
        .executableTarget(name: "App", dependencies: [
            // Other dependencies
            .product(name: "QueuesRedisDriver", package: "queues-redis-driver")
        ]),
        .testTarget(name: "AppTests", dependencies: [.target(name: "App")]),
    ]
)

Jeśli edytujesz manifest bezpośrednio w Xcode, automatycznie wykryje on zmiany i pobierze nową zależność po zapisaniu pliku. W przeciwnym razie z terminala uruchom swift package resolve, aby pobrać nową zależność.

Konfiguracja#

Kolejnym krokiem jest skonfigurowanie Queues w configure.swift. Jako przykład użyjemy biblioteki Redis:

import QueuesRedisDriver

try app.queues.use(.redis(url: "redis://127.0.0.1:6379"))

Rejestrowanie Joba#

Po zamodelowaniu zadania musisz dodać je do sekcji konfiguracji w następujący sposób:

// Register jobs
let emailJob = EmailJob()
app.queues.add(emailJob)

Uruchamianie workerów jako procesów#

Aby uruchomić nowego workera kolejki, wykonaj swift run App queues. Możesz również wskazać konkretny typ workera do uruchomienia: swift run App queues --queue emails.

Uruchamianie workerów w tym samym procesie#

Aby uruchomić workera w tym samym procesie co Twoja aplikacja (zamiast uruchamiać do tego celu całkowicie osobny serwer), wywołaj metody pomocnicze na Application:

try app.queues.startInProcessJobs(on: .default)

Aby uruchomić zaplanowane zadania w tym samym procesie, wywołaj następującą metodę:

try app.queues.startScheduledJobs()

Protokół Job#

Zadania są definiowane za pomocą protokołu Job lub AsyncJob.

Modelowanie obiektu Job:#

import Vapor
import Foundation
import Queues

struct Email: Codable {
    let to: String
    let message: String
}

struct EmailJob: Job {
    typealias Payload = Email
    
    func dequeue(_ context: QueueContext, _ payload: Email) -> EventLoopFuture<Void> {
        // This is where you would send the email
        return context.eventLoop.future()
    }
    
    func error(_ context: QueueContext, _ error: Error, _ payload: Email) -> EventLoopFuture<Void> {
        // If you don't want to handle errors you can simply return a future. You can also omit this function entirely.
        return context.eventLoop.future()
    }
}

Jeśli używasz async/await, powinieneś użyć AsyncJob:

struct EmailJob: AsyncJob {
    typealias Payload = Email
    
    func dequeue(_ context: QueueContext, _ payload: Email) async throws {
        // This is where you would send the email
    }
    
    func error(_ context: QueueContext, _ error: Error, _ payload: Email) async throws {
        // If you don't want to handle errors you can simply return. You can also omit this function entirely.
    }
}

Wysyłanie zadań#

Aby wysłać zadanie do kolejki, potrzebujesz dostępu do instancji Application lub Request. Najczęściej będziesz wysyłać zadania wewnątrz handlera trasy:

app.get("email") { req -> EventLoopFuture<String> in
    return req
        .queue
        .dispatch(
            EmailJob.self,
            .init(to: "email@email.com", message: "message")
        ).map { "done" }
}

// or

app.get("email") { req async throws -> String in
    try await req.queue.dispatch(
        EmailJob.self,
        .init(to: "email@email.com", message: "message"))
    return "done"
}

Jeśli natomiast musisz wysłać zadanie z kontekstu, w którym obiekt Request nie jest dostępny (na przykład wewnątrz Command), musisz skorzystać z właściwości queues wewnątrz obiektu Application, na przykład tak:

struct SendEmailCommand: AsyncCommand {
    func run(using context: CommandContext, signature: Signature) async throws {
        context
            .application
            .queues
            .queue
            .dispatch(
                EmailJob.self,
                .init(to: "email@email.com", message: "message")
            )
    }
}

Ustawianie maxRetryCount#

Zadania będą automatycznie ponawiane w przypadku błędu, jeśli określisz maxRetryCount. Na przykład:

app.get("email") { req -> EventLoopFuture<String> in
    return req
        .queue
        .dispatch(
            EmailJob.self,
            .init(to: "email@email.com", message: "message"),
            maxRetryCount: 3
        ).map { "done" }
}

// or

app.get("email") { req async throws -> String in
    try await req.queue.dispatch(
        EmailJob.self,
        .init(to: "email@email.com", message: "message"),
        maxRetryCount: 3)
    return "done"
}

Określanie opóźnienia#

Zadania mogą być również ustawione tak, aby uruchamiały się dopiero po upływie określonej Date. Aby określić opóźnienie, przekaż Date do parametru delayUntil w dispatch:

app.get("email") { req async throws -> String in
    let futureDate = Date(timeIntervalSinceNow: 60 * 60 * 24) // One day
    try await req.queue.dispatch(
        EmailJob.self,
        .init(to: "email@email.com", message: "message"),
        maxRetryCount: 3,
        delayUntil: futureDate)
    return "done"
}

Jeśli zadanie zostanie pobrane z kolejki przed upływem parametru opóźnienia, zostanie ono ponownie umieszczone w kolejce przez driver.

Określanie priorytetu#

Zadania mogą być sortowane do różnych typów/priorytetów kolejek, w zależności od Twoich potrzeb. Możesz na przykład chcieć otworzyć kolejkę email oraz kolejkę background-processing, aby posortować zadania.

Zacznij od rozszerzenia QueueName:

extension QueueName {
    static let emails = QueueName(string: "emails")
}

Możesz również ustawić workerCount dla poszczególnej kolejki podczas tworzenia QueueName:

extension QueueName {
    static let serialEmails = QueueName(string: "serial-emails", workerCount: 1)
}

Ustawienie workerCount: 1 sprawia, że dana kolejka przetwarza zadania kolejno, co jest przydatne, gdy kolejność zadań ma znaczenie.

Następnie określ typ kolejki podczas pobierania obiektu jobs:

app.get("email") { req -> EventLoopFuture<String> in
    let futureDate = Date(timeIntervalSinceNow: 60 * 60 * 24) // One day
    return req
        .queues(.emails)
        .dispatch(
            EmailJob.self,
            .init(to: "email@email.com", message: "message"),
            maxRetryCount: 3,
            delayUntil: futureDate
        ).map { "done" }
}

// or

app.get("email") { req async throws -> String in
    let futureDate = Date(timeIntervalSinceNow: 60 * 60 * 24) // One day
    try await req
        .queues(.emails)
        .dispatch(
            EmailJob.self,
            .init(to: "email@email.com", message: "message"),
            maxRetryCount: 3,
            delayUntil: futureDate
        )
    return "done"
}

Podczas dostępu z poziomu obiektu Application powinieneś postąpić następująco:

struct SendEmailCommand: AsyncCommand {
    func run(using context: CommandContext, signature: Signature) async throws {
        context
            .application
            .queues
            .queue(.emails)
            .dispatch(
                EmailJob.self,
                .init(to: "email@email.com", message: "message"),
                maxRetryCount: 3,
                delayUntil: futureDate
            )
    }
}

Jeśli nie określisz kolejki, zadanie zostanie uruchomione na kolejce default. Upewnij się, że postępujesz zgodnie z instrukcjami w sekcji Pierwsze kroki, aby uruchomić workery dla każdego typu kolejki.

Planowanie zadań#

Pakiet Queues pozwala również na planowanie zadań do wykonania w określonych momentach czasu.

Uruchamianie workera schedulera#

Scheduler wymaga uruchomienia osobnego procesu workera, podobnie jak worker kolejki. Możesz uruchomić workera, wykonując to polecenie:

swift run App queues --scheduled

Tworzenie ScheduledJoba#

Na początek utwórz nowy ScheduledJob lub AsyncScheduledJob:

import Vapor
import Queues

struct CleanupJob: ScheduledJob {
    // Add extra services here via dependency injection, if you need them.

    func run(context: QueueContext) -> EventLoopFuture<Void> {
        // Do some work here, perhaps queue up another job.
        return context.eventLoop.makeSucceededFuture(())
    }
}

struct CleanupJob: AsyncScheduledJob {
    // Add extra services here via dependency injection, if you need them.

    func run(context: QueueContext) async throws {
        // Do some work here, perhaps queue up another job.
    }
}

Następnie w kodzie konfiguracji zarejestruj zaplanowane zadanie:

app.queues.schedule(CleanupJob())
    .yearly()
    .in(.may)
    .on(23)
    .at(.noon)

Zadanie z powyższego przykładu będzie uruchamiane co roku, 23 maja o godzinie 12:00.

Dostępne metody buildera#

Istnieją dwa style API schedulera:

• Buildery w stylu kalendarza, które zwracają obiekty buildera do dalszego łańcuchowania.

• Buildery w stylu interwałowym, które uruchamiają zadania co ustalony odstęp czasu.

Powinieneś kontynuować budowanie łańcucha schedulera w stylu kalendarza, dopóki kompilator nie przestanie ostrzegać o nieużywanym wyniku. Poniżej znajdują się wszystkie dostępne metody:

Metody buildera interwałowego (.every(...))#

Scheduler obsługuje również planowanie w stałych odstępach czasu za pomocą metod .every(...):

Przykład:

app.queues.schedule(CleanupJob())
    .every(hours: 6)

Dostępne pomocniki#

Queues zawiera kilka pomocniczych enumów, które ułatwiają planowanie:

Aby użyć pomocniczego enuma, wywołaj odpowiedni modyfikator na funkcji pomocniczej i przekaż wartość. Na przykład:

// Every year in January
.yearly().in(.january)

// Every month on the first day
.monthly().on(.first)

// Every week on Sunday
.weekly().on(.sunday)

// Every day at midnight
.daily().at(.midnight)

Delegaci zdarzeń#

Pakiet Queues pozwala określić obiekty JobEventDelegate, które będą otrzymywać powiadomienia, gdy worker podejmie działanie na zadaniu. Może to być wykorzystane do monitorowania, wyciągania wniosków lub celów związanych z alertowaniem.

Na początek dostosuj obiekt do protokołu JobEventDelegate i zaimplementuj wymagane metody

struct MyEventDelegate: JobEventDelegate {
    /// Called when the job is dispatched to the queue worker from a route
    func dispatched(job: JobEventData, eventLoop: EventLoop) -> EventLoopFuture<Void> {
        eventLoop.future()
    }

    /// Called when the job is placed in the processing queue and work begins
    func didDequeue(jobId: String, eventLoop: EventLoop) -> EventLoopFuture<Void> {
        eventLoop.future()
    }

    /// Called when the job has finished processing and has been removed from the queue
    func success(jobId: String, eventLoop: EventLoop) -> EventLoopFuture<Void> {
        eventLoop.future()
    }

    /// Called when the job has finished processing but had an error
    func error(jobId: String, error: Error, eventLoop: EventLoop) -> EventLoopFuture<Void> {
        eventLoop.future()
    }
}

Następnie dodaj go w pliku konfiguracyjnym:

app.queues.add(MyEventDelegate())

Istnieje kilka pakietów firm trzecich, które wykorzystują funkcjonalność delegatów, aby dostarczyć dodatkowy wgląd w działanie Twoich workerów kolejki:

• QueuesDatabaseHooks

• QueuesDash

Testowanie#

Aby uniknąć problemów z synchronizacją i zapewnić deterministyczne testowanie, pakiet Queues udostępnia bibliotekę XCTQueue oraz dedykowany do testów driver AsyncTestQueuesDriver, którego możesz użyć w następujący sposób:

final class UserCreationServiceTests: XCTestCase {
    var app: Application!

    override func setUp() async throws {
        self.app = try await Application.make(.testing)
        try await configure(app)

        // Override the driver being used for testing
        app.queues.use(.asyncTest)
    }

    override func tearDown() async throws {
        try await self.app.asyncShutdown()
        self.app = nil
    }
}

Zobacz więcej szczegółów we wpisie na blogu Romaina Pouclet.

Rozwiązywanie problemów#

Podczas korzystania z queues-redis-driver z serwerem kompatybilnym z Redis działającym w trybie klastrowym, takim jak Redis lub Valkey na Amazon AWS, możesz napotkać ten komunikat błędu: CROSSSLOT Keys in request don't hash to the same slot.

Zdarza się to wyłącznie w trybie klastrowym, ponieważ Redis lub Valkey nie mogą mieć pewności, na którym węźle klastra przechowywać dane zadania.

Aby to naprawić, dodaj hash tag do nazw wpisów danych Twoich zadań, używając nawiasów klamrowych w nazwach:

app.queues.configuration.persistenceKey = "vapor-queues-{queues}"