Vodiči

Build a Resilient PHP 8.3 CLI Worker for Production Workloads

Izradite otporan PHP 8.3 CLI radnik za produkcijska opterećenja

Radnik reda je jednostavan sve do prvog nezgodnog rušenja. U idealnom slučaju dohvaća redak i poziva obrađivač. Produkcija dodaje konkurentske procese, otrovne poruke, neizvjesna predavanja, istekle rezervacije, implementacije i signale za prekid koji stižu točno na pogrešnoj granici.

Ovaj vodič izrađuje izvorni PHP 8.3 radnik s PDO-om i MySQL-om 8. Koristi FOR UPDATE SKIP LOCKED za istodobno preuzimanje, potvrđuje rezervacije prije obrade, oporavlja istekle najmove, primjenjuje ograničene ponovne pokušaje, premješta konačne neuspjehe u mrtvo slovo i gasi se putem pcntl.

Njegovo jamstvo isporuke je najmanje jednom. Obrađivač se može ponovno pokrenuti nakon rušenja ili isteka najma. Pouzdanost stoga ovisi o zaštiti poslovnog učinka stabilnim idempotentnim ključem tamo gdje se taj učinak potvrđuje.

Preduvjeti i arhitektonske granice

Trebate PHP 8.3 CLI s pdo_mysql, mysqlnd i pcntl; MySQL 8 koji koristi InnoDB; te Linux poslužitelj za dugotrajni proces.

php --version
php -m | grep -E '^(PDO|pdo_mysql|mysqlnd|pcntl)$'
mysql --version

Radnik prati tri granice:

  1. Preuzmite jedan prihvatljivi redak u kratkoj transakciji, dodijelite nasumični token najma i potvrdite transakciju.
  2. Pokrenite poslovnu logiku bez zadržavanja zaključavanja redaka.
  3. U drugoj kratkoj transakciji provjerite vlasništvo, zapišite idempotentni poslovni učinak i označite zadatak dovršenim.

Najam je nadoknadiva tvrdnja o vlasništvu, a ne transakcija koja se zadržava tijekom cijelog izvršavanja. Ako proces nestane, postupak oporavka vraća istekle rezervacije u red ili ih označava mrtvima kada više nema pokušaja.

Struktura projekta

report-worker/
├── database/
│   └── schema.sql
├── src/
│   └── config.php
└── bin/
    ├── enqueue.php
    └── worker.php

Izradite tablice reda i učinka

Primijenite database/schema.sql s administrativnim MySQL identitetom. Ovaj primjer zadržava MySQL na istom poslužitelju; host računa prilagodite samo pri implementaciji preko privatne mreže.

CREATE DATABASE report_queue
    CHARACTER SET utf8mb4
    COLLATE utf8mb4_0900_ai_ci;

USE report_queue;

CREATE TABLE jobs (
    id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
    queue_name VARCHAR(64) NOT NULL,
    job_type VARCHAR(100) NOT NULL,
    payload JSON NOT NULL,
    idempotency_key CHAR(64) NOT NULL,
    status ENUM('ready', 'reserved', 'done', 'dead')
        NOT NULL DEFAULT 'ready',
    attempts SMALLINT UNSIGNED NOT NULL DEFAULT 0,
    max_attempts SMALLINT UNSIGNED NOT NULL DEFAULT 5,
    available_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
    lease_until DATETIME(6) NULL,
    lease_token BINARY(16) NULL,
    last_error VARCHAR(4000) NULL,
    created_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
    updated_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6)
        ON UPDATE CURRENT_TIMESTAMP(6),
    PRIMARY KEY (id),
    UNIQUE KEY uq_job_idempotency (queue_name, idempotency_key),
    KEY ix_claim (queue_name, status, available_at, id),
    KEY ix_recovery (queue_name, status, lease_until, id)
) ENGINE=InnoDB;

CREATE TABLE generated_reports (
    id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT,
    idempotency_key CHAR(64) NOT NULL,
    tenant_id VARCHAR(100) NOT NULL,
    report_month CHAR(7) NOT NULL,
    result_json JSON NOT NULL,
    created_at DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
    PRIMARY KEY (id),
    UNIQUE KEY uq_report_idempotency (idempotency_key)
) ENGINE=InnoDB;

CREATE USER 'queue_worker'@'127.0.0.1'
    IDENTIFIED BY 'replace-with-a-generated-secret';

GRANT SELECT, INSERT, UPDATE
    ON report_queue.*
    TO 'queue_worker'@'127.0.0.1';

Prvi jedinstveni ključ sprječava dvostruke zahtjeve za stavljanje u red. Drugi štiti stvarni učinak u bazi podataka ako se obrada ponovi. Identitet tijekom izvođenja nema ovlasti za upravljanje shemom ni brisanje.

Ograničite veze, mrežna čitanja i čekanja zaključavanja

mysqlnd.net_read_timeout je PHP postavka na razini sustava, stoga je konfigurirajte prije pokretanja PHP-a. Najprije pronađite CLI konfiguraciju pomoću php --ini. Ovaj primjer za Debian/Ubuntu dodaje namjensko CLI nadjačavanje; na drugoj distribuciji dodajte istu direktivu u učitani CLI php.ini ili njegov skenirani konfiguracijski direktorij.

php --ini
sudo install -d -m 0755 /etc/php/8.3/cli/conf.d
printf '%s\n' 'mysqlnd.net_read_timeout=10' \
  | sudo tee /etc/php/8.3/cli/conf.d/99-report-worker.ini >/dev/null
php -r 'echo ini_get("mysqlnd.net_read_timeout"), PHP_EOL;'

Posljednja naredba mora ispisati 10. Izradite src/config.php. Vremenska ograničenja služe različitim svrhama i namjerno su kraća od najma od 60 sekundi.

<?php
declare(strict_types=1);

function database(): PDO
{
    static $pdo = null;

    if ($pdo instanceof PDO) {
        return $pdo;
    }

    if (!extension_loaded('mysqlnd')) {
        throw new RuntimeException('mysqlnd is required');
    }

    if ((int) ini_get('mysqlnd.net_read_timeout') !== 10) {
        throw new RuntimeException(
            'Set mysqlnd.net_read_timeout=10 in the CLI php.ini'
        );
    }

    $password = getenv('DB_PASS');
    if ($password === false || $password === '') {
        throw new RuntimeException('DB_PASS is required');
    }

    $host = getenv('DB_HOST') ?: '127.0.0.1';
    $port = getenv('DB_PORT') ?: '3306';
    $name = getenv('DB_NAME') ?: 'report_queue';
    $user = getenv('DB_USER') ?: 'queue_worker';

    $dsn = "mysql:host={$host};port={$port};dbname={$name};charset=utf8mb4";

    $pdo = new PDO($dsn, $user, $password, [
        PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION,
        PDO::ATTR_DEFAULT_FETCH_MODE => PDO::FETCH_ASSOC,
        PDO::ATTR_EMULATE_PREPARES => false,
        PDO::ATTR_TIMEOUT => 5,
        PDO::MYSQL_ATTR_MULTI_STATEMENTS => false,
    ]);

    $pdo->exec("SET SESSION time_zone = '+00:00'");
    $pdo->exec("SET SESSION innodb_lock_wait_timeout = 3");
    $pdo->exec("SET SESSION lock_wait_timeout = 5");

    return $pdo;
}

PDO::ATTR_TIMEOUT ograničava uspostavljanje veze za PDO MySQL; nije rok za upite. mysqlnd.net_read_timeout ograničava pojedinačno mrežno čitanje na deset sekundi. I dalje ne otkazuje rad na strani poslužitelja, stoga radnik izlazi nakon neočekivane iznimke baze podataka i prepušta nadzorniku stvaranje čiste veze.

innodb_lock_wait_timeout ograničava čekanja zaključavanja InnoDB zapisa, dok lock_wait_timeout pokriva zaključavanja metapodataka. Ove postavke ne pretvaraju zahtjev za preuzimanje s zaključavanjem SELECT ... FOR UPDATE u strogi rok za upit. Klijentski mysqlnd.net_read_timeout sprječava da pojedinačno mrežno čitanje čeka zauvijek; nakon neočekivanog isteka vremena baze podataka, radnik izlazi, a njegov nadzornik stvara čistu vezu. Neka svaki upit za preuzimanje koristi indeks, a svaka transakcija rezervacije bude kratka.

Idempotentno stavljajte zadatke u red

Izradite bin/enqueue.php. Njegov ključ potječe iz poslovnog identiteta mjesečnog izvješća, tako da ponavljanje naredbe vraća postojeći ID zadatka.

<?php
declare(strict_types=1);

require dirname(__DIR__) . '/src/config.php';

if ($argc !== 3) {
    fwrite(STDERR, "Usage: php bin/enqueue.php TENANT_ID YYYY-MM\n");
    exit(64);
}

$tenant = $argv[1];
$month = $argv[2];

if (!preg_match('/^[A-Za-z0-9_-]{1,100}$/D', $tenant)) {
    throw new InvalidArgumentException('Invalid tenant ID');
}

if (!preg_match('/^\d{4}-(0[1-9]|1[0-2])$/D', $month)) {
    throw new InvalidArgumentException('Month must use YYYY-MM');
}

$key = hash('sha256', "monthly-report:{$tenant}:{$month}");
$payload = json_encode([
    'tenant_id' => $tenant,
    'report_month' => $month,
], JSON_THROW_ON_ERROR);

$sql = <<<'SQL'
INSERT INTO jobs (queue_name, job_type, payload, idempotency_key)
VALUES ('reports', 'report.generate', ?, ?)
ON DUPLICATE KEY UPDATE id = LAST_INSERT_ID(id)
SQL;

$pdo = database();
$statement = $pdo->prepare($sql);
$statement->execute([$payload, $key]);

fwrite(STDOUT, $pdo->lastInsertId() . PHP_EOL);

Implementirajte preuzimanja, ponovne pokušaje i elegantno oslobađanje

Izradite bin/worker.php. Oporavak obrađuje najviše 100 redaka po prolazu. Indeksirano preuzimanje koristi SKIP LOCKED i kratku transakciju. PDO MySQL nema univerzalni rok po naredbi, stoga radnik kombinira ograničenja čekanja zaključavanja na poslužitelju s ograničenim klijentskim vremenskim ograničenjem za mrežno čitanje i ponovnim pokretanjem nadzornika.

<?php
declare(strict_types=1);

require dirname(__DIR__) . '/src/config.php';

const QUEUE = 'reports';
const LEASE_SECONDS = 60;

function recoverExpiredLeases(PDO $pdo): int
{
    $sql = <<<'SQL'
UPDATE jobs
SET status = CASE
        WHEN attempts >= max_attempts THEN 'dead'
        ELSE 'ready'
    END,
    available_at = CASE
        WHEN attempts >= max_attempts THEN available_at
        ELSE UTC_TIMESTAMP(6)
    END,
    lease_until = NULL,
    lease_token = NULL,
    last_error = CASE
        WHEN attempts >= max_attempts
            THEN 'Lease expired after final attempt'
        ELSE 'Lease expired; returned to queue'
    END
WHERE queue_name = ?
  AND status = 'reserved'
  AND lease_until <= UTC_TIMESTAMP(6)
ORDER BY lease_until, id
LIMIT 100
SQL;

    $statement = $pdo->prepare($sql);
    $statement->execute([QUEUE]);
    return $statement->rowCount();
}

function claim(PDO $pdo): ?array
{
    $pdo->beginTransaction();

    try {
        $select = $pdo->prepare(
            "SELECT *
             FROM jobs
             WHERE queue_name = ?
               AND status = 'ready'
               AND attempts < max_attempts
               AND available_at <= UTC_TIMESTAMP(6)
             ORDER BY available_at, id
             LIMIT 1
             FOR UPDATE SKIP LOCKED"
        );
        $select->execute([QUEUE]);
        $job = $select->fetch();

        if ($job === false) {
            $pdo->commit();
            return null;
        }

        $token = random_bytes(16);
        $update = $pdo->prepare(
            "UPDATE jobs
             SET status = 'reserved',
                 attempts = attempts + 1,
                 lease_token = ?,
                 lease_until = TIMESTAMPADD(
                     SECOND, ?, UTC_TIMESTAMP(6)
                 )
             WHERE id = ?"
        );
        $update->execute([$token, LEASE_SECONDS, $job['id']]);
        $pdo->commit();

        $job['attempts'] = (int) $job['attempts'] + 1;
        $job['lease_token'] = $token;

        $testPause = (int) (getenv('WORKER_TEST_POST_CLAIM_PAUSE_MS') ?: 0);
        if ($testPause > 0 && $testPause <= 10_000) {
            usleep($testPause * 1000);
        }

        return $job;
    } catch (Throwable $error) {
        if ($pdo->inTransaction()) {
            $pdo->rollBack();
        }
        throw $error;
    }
}

function releaseClaim(PDO $pdo, array $job): bool
{
    $statement = $pdo->prepare(
        "UPDATE jobs
         SET status = 'ready',
             attempts = IF(attempts > 0, attempts - 1, 0),
             available_at = UTC_TIMESTAMP(6),
             lease_until = NULL,
             lease_token = NULL,
             last_error = 'Released during shutdown before processing'
         WHERE id = ?
           AND status = 'reserved'
           AND lease_token = ?"
    );
    $statement->execute([$job['id'], $job['lease_token']]);

    return $statement->rowCount() === 1;
}

function processJob(array $job): string
{
    if ($job['job_type'] !== 'report.generate') {
        throw new DomainException('Unsupported job type');
    }

    $payload = json_decode(
        $job['payload'],
        true,
        flags: JSON_THROW_ON_ERROR
    );

    $tenant = $payload['tenant_id'] ?? null;
    $month = $payload['report_month'] ?? null;

    if (
        !is_string($tenant)
        || !preg_match('/^[A-Za-z0-9_-]{1,100}$/D', $tenant)
        || !is_string($month)
        || !preg_match('/^\d{4}-(0[1-9]|1[0-2])$/D', $month)
    ) {
        throw new UnexpectedValueException('Malformed job payload');
    }

    if (getenv('WORKER_ENABLE_TEST_FIXTURES') === '1') {
        if (!empty($payload['test_pause'])) {
            $until = hrtime(true) + 30_000_000_000;
            while (hrtime(true) < $until) {
                sleep(1);
            }
        }

        $transientUntil = (int) (
            $payload['test_transient_until_attempt'] ?? 0
        );
        if (
            $transientUntil > 0
            && (int) $job['attempts'] <= $transientUntil
        ) {
            throw new RuntimeException(
                'Synthetic transient test failure'
            );
        }
    }

    return json_encode([
        'tenant_id' => $tenant,
        'report_month' => $month,
        'state' => 'generated',
    ], JSON_THROW_ON_ERROR);
}

function ownsLease(array|false $row, string $token): bool
{
    return $row !== false
        && (int) $row['lease_valid'] === 1
        && hash_equals($row['lease_token'], $token);
}

function complete(PDO $pdo, array $job, string $result): bool
{
    $pdo->beginTransaction();

    try {
        $lock = $pdo->prepare(
            "SELECT lease_token,
                    lease_until > UTC_TIMESTAMP(6) AS lease_valid
             FROM jobs
             WHERE id = ? AND status = 'reserved'
             FOR UPDATE"
        );
        $lock->execute([$job['id']]);
        $current = $lock->fetch();

        if (!ownsLease($current, $job['lease_token'])) {
            $pdo->rollBack();
            return false;
        }

        $payload = json_decode(
            $job['payload'],
            true,
            flags: JSON_THROW_ON_ERROR
        );

        $effect = $pdo->prepare(
            "INSERT INTO generated_reports (
                idempotency_key, tenant_id, report_month, result_json
             ) VALUES (?, ?, ?, ?)
             ON DUPLICATE KEY UPDATE idempotency_key = ?"
        );
        $effect->execute([
            $job['idempotency_key'],
            $payload['tenant_id'],
            $payload['report_month'],
            $result,
            $job['idempotency_key'],
        ]);

        $finish = $pdo->prepare(
            "UPDATE jobs
             SET status = 'done',
                 lease_until = NULL,
                 lease_token = NULL,
                 last_error = NULL
             WHERE id = ?"
        );
        $finish->execute([$job['id']]);

        $pdo->commit();
        return true;
    } catch (Throwable $error) {
        if ($pdo->inTransaction()) {
            $pdo->rollBack();
        }
        throw $error;
    }
}

function errorSummary(Throwable $error): string
{
    $controlled = $error instanceof DomainException
        || $error instanceof UnexpectedValueException
        || $error instanceof JsonException;

    return $controlled
        ? substr($error::class . ': ' . $error->getMessage(), 0, 4000)
        : $error::class;
}

function failJob(PDO $pdo, array $job, Throwable $error): bool
{
    $pdo->beginTransaction();

    try {
        $lock = $pdo->prepare(
            "SELECT attempts, max_attempts, lease_token,
                    lease_until > UTC_TIMESTAMP(6) AS lease_valid
             FROM jobs
             WHERE id = ? AND status = 'reserved'
             FOR UPDATE"
        );
        $lock->execute([$job['id']]);
        $current = $lock->fetch();

        if (!ownsLease($current, $job['lease_token'])) {
            $pdo->rollBack();
            return false;
        }

        $permanent = $error instanceof DomainException
            || $error instanceof UnexpectedValueException
            || $error instanceof JsonException;
        $exhausted = $permanent
            || (int) $current['attempts']
                >= (int) $current['max_attempts'];
        $summary = errorSummary($error);

        if ($exhausted) {
            $statement = $pdo->prepare(
                "UPDATE jobs
                 SET status = 'dead',
                     lease_until = NULL,
                     lease_token = NULL,
                     last_error = ?
                 WHERE id = ?"
            );
            $statement->execute([$summary, $job['id']]);
        } else {
            $attempt = (int) $current['attempts'];
            $delay = min(
                300,
                (2 ** min($attempt, 8)) + random_int(0, 3)
            );

            $statement = $pdo->prepare(
                "UPDATE jobs
                 SET status = 'ready',
                     available_at = TIMESTAMPADD(
                         SECOND, ?, UTC_TIMESTAMP(6)
                     ),
                     lease_until = NULL,
                     lease_token = NULL,
                     last_error = ?
                 WHERE id = ?"
            );
            $statement->execute([$delay, $summary, $job['id']]);
        }

        $pdo->commit();
        return true;
    } catch (Throwable $failure) {
        if ($pdo->inTransaction()) {
            $pdo->rollBack();
        }
        throw $failure;
    }
}

function logEvent(string $event, array $context = []): void
{
    fwrite(STDERR, json_encode([
        'event' => $event,
        'time' => gmdate(DATE_ATOM),
    ] + $context, JSON_THROW_ON_ERROR) . PHP_EOL);
}

$stopping = false;
pcntl_async_signals(true);

$stop = static function (int $signal) use (&$stopping): void {
    $stopping = true;
};

pcntl_signal(SIGTERM, $stop);
pcntl_signal(SIGINT, $stop);

try {
    $pdo = database();
    $nextRecoveryAt = 0.0;
    logEvent('worker_started', ['pid' => getmypid()]);

    while (!$stopping) {
        if (microtime(true) >= $nextRecoveryAt) {
            $recovered = recoverExpiredLeases($pdo);
            $nextRecoveryAt = microtime(true) + 5.0;

            if ($recovered > 0) {
                logEvent('leases_recovered', ['count' => $recovered]);
            }
        }

        if ($stopping) {
            break;
        }

        $job = claim($pdo);

        /*
         * This check must be the first action after claim().
         * No handler starts if shutdown arrived during reservation.
         */
        if ($stopping) {
            if ($job !== null) {
                try {
                    $released = releaseClaim($pdo, $job);
                    logEvent(
                        $released
                            ? 'claim_released_on_shutdown'
                            : 'claim_release_lost',
                        ['job_id' => (int) $job['id']]
                    );
                } catch (Throwable $releaseError) {
                    logEvent('claim_release_failed', [
                        'job_id' => (int) $job['id'],
                        'error' => errorSummary($releaseError),
                    ]);
                    throw $releaseError;
                }
            }
            break;
        }

        if ($job === null) {
            for ($tick = 0; $tick < 10 && !$stopping; $tick++) {
                usleep(250_000);
            }
            continue;
        }

        $startedAt = hrtime(true);
        logEvent('job_claimed', [
            'job_id' => (int) $job['id'],
            'attempt' => $job['attempts'],
        ]);

        try {
            $result = processJob($job);
        } catch (Throwable $handlerError) {
            $owned = failJob($pdo, $job, $handlerError);
            logEvent(
                $owned ? 'job_failed' : 'job_lease_lost',
                [
                    'job_id' => (int) $job['id'],
                    'error' => errorSummary($handlerError),
                    'elapsed_ms' => (int) (
                        (hrtime(true) - $startedAt) / 1_000_000
                    ),
                ]
            );
            continue;
        }

        /*
         * Do not classify finalization failures as handler failures.
         * A database exception here escapes to the outer fatal handler,
         * so systemd restarts the worker with a fresh connection.
         */
        $owned = complete($pdo, $job, $result);
        logEvent(
            $owned ? 'job_completed' : 'job_lease_lost',
            [
                'job_id' => (int) $job['id'],
                'elapsed_ms' => (int) (
                    (hrtime(true) - $startedAt) / 1_000_000
                ),
            ]
        );
    }

    logEvent('worker_stopped', ['pid' => getmypid()]);
} catch (Throwable $fatal) {
    logEvent('worker_fatal', ['error' => errorSummary($fatal)]);
    exit(1);
}

SKIP LOCKED sprječava radnike da čekaju iza retka koji je odabrao drugi radnik, ali transakcija i dalje ostaje ključna: redak ostaje zaključan dok se ne pohrane njegov token najma i rok.

Neposredna provjera zaustavljanja zatvara suptilnu utrku pri gašenju. Ako SIGTERM stigne dok je claim() blokiran ili pri završetku, radnik uvjetno ažurira samo redak koji odgovara i ID-u zadatka i nasumičnom tokenu najma. Vraća taj redak u ready, vraća neiskorišteni pokušaj, bilježi je li oslobađanje uspjelo i nikada ne poziva processJob().

Poslovni učinak i prijelaz u done dijele jednu transakciju. Za vanjski API pošaljite stabilni idempotentni ključ kada je podržan. MySQL ne može atomski koordinirati nepovezani udaljeni sporedni učinak; u suprotnom koristite usklađivanje nizvodno ili integraciju temeljenu na outboxu.

Samo iznimke koje podigne processJob() ulaze u failJob() i troše pokušaj. Neuspjeh u complete() ili u transakciji stanja neuspjeha infrastrukturni je neuspjeh: dolazi do vanjskog obrađivača fatalnih pogrešaka, izlazi iz procesa i dopušta systemdu ponovno pokretanje sa svježom vezom. Potvrđeni najam tada sprječava trenutačno dvostruko vlasništvo i ostaje nadoknadiv nakon isteka.

Isprobajte konkurentnost i putanje neuspjeha

Izvezite vjerodajnice u svakom terminalu, dvaput stavite isto izvješće u red i pokrenite dva radnika:

export DB_HOST=127.0.0.1
export DB_PORT=3306
export DB_NAME=report_queue
export DB_USER=queue_worker
export DB_PASS='replace-with-a-generated-secret'

php bin/enqueue.php acme 2042-01
php bin/enqueue.php acme 2042-01
php bin/worker.php

Naredbe za stavljanje u red trebale bi ispisati isti ID. Dodajte nekoliko klijenata i potvrdite da zasebni radnici preuzimaju različite retke. Pregledajte trajno stanje pomoću:

SELECT id, status, attempts, available_at, lease_until, last_error
FROM jobs
ORDER BY id DESC
LIMIT 20;

SELECT idempotency_key, tenant_id, report_month, result_json
FROM generated_reports
ORDER BY id DESC
LIMIT 20;

Testirajte putanju gašenja nakon preuzimanja na razvojnoj instanci. Pauza koja je prema zadanim postavkama onemogućena čini vremenski slijed determinističkim:

php bin/enqueue.php shutdown-test 2042-02

WORKER_TEST_POST_CLAIM_PAUSE_MS=10000 php bin/worker.php &
worker_pid=$!

case "$worker_pid" in
    ''|*[!0-9]*) exit 1 ;;
esac

sleep 1
kill -TERM "$worker_pid"
wait "$worker_pid"

Dnevnik bi trebao sadržavati claim_released_on_shutdown, nakon čega slijedi worker_stopped. Zadatak bi trebao biti ready, bez generiranog izvješća i bez potrošenog pokušaja.

Preostale pomoćne scenarije za neuspjehe pokrenite samo nad izoliranom razvojnom bazom podataka. Ponašanje obrađivača samo za testiranje onemogućeno je osim ako nije postavljeno WORKER_ENABLE_TEST_FIXTURES=1.

Rušenje i oporavak isteklog najma

pause_id=$(
  MYSQL_PWD="$DB_PASS" mysql --protocol=TCP     -h "$DB_HOST" -P "$DB_PORT" -u "$DB_USER" "$DB_NAME" -Nse "
      INSERT INTO jobs (
        queue_name, job_type, payload, idempotency_key, max_attempts, available_at
      ) VALUES (
        'reports',
        'report.generate',
        JSON_OBJECT(
          'tenant_id', 'fixture-pause',
          'report_month', '2042-03',
          'test_pause', TRUE
        ),
        SHA2(CONCAT('fixture:pause:', UUID()), 256),
        3,
        UTC_TIMESTAMP(6)
      );
      SELECT LAST_INSERT_ID();
    "
)

WORKER_ENABLE_TEST_FIXTURES=1 php bin/worker.php   2> /tmp/report-worker-pause.log &
worker_pid=$!

for attempt in $(seq 1 50); do
  grep -q '"event":"job_claimed"' /tmp/report-worker-pause.log     && break
  sleep 0.1
done
grep -q '"event":"job_claimed"' /tmp/report-worker-pause.log
kill -KILL "$worker_pid"
wait "$worker_pid" || true

sleep 61
timeout --signal=TERM 8s php bin/worker.php   2>> /tmp/report-worker-pause.log || test "$?" -eq 124
grep -q '"event":"leases_recovered"' /tmp/report-worker-pause.log

MYSQL_PWD="$DB_PASS" mysql --protocol=TCP   -h "$DB_HOST" -P "$DB_PORT" -u "$DB_USER" "$DB_NAME"   -e "SELECT id, status, attempts FROM jobs WHERE id = $pause_id"

Završni upit mora prikazati done. Dnevnik mora sadržavati i job_claimed i leases_recovered, čime se dokazuje da je drugi radnik ponovno preuzeo istekli najam od 60 sekundi.

Trajni i prolazni neuspjesi

unsupported_id=$(
  MYSQL_PWD="$DB_PASS" mysql --protocol=TCP     -h "$DB_HOST" -P "$DB_PORT" -u "$DB_USER" "$DB_NAME" -Nse "
      INSERT INTO jobs (
        queue_name, job_type, payload, idempotency_key, max_attempts, available_at
      ) VALUES (
        'reports',
        'unsupported.fixture',
        JSON_OBJECT(
          'tenant_id', 'fixture-unsupported',
          'report_month', '2042-04'
        ),
        SHA2(CONCAT('fixture:unsupported:', UUID()), 256),
        1,
        UTC_TIMESTAMP(6)
      );
      SELECT LAST_INSERT_ID();
    "
)
timeout --signal=TERM 5s php bin/worker.php   2> /tmp/report-worker-permanent.log || test "$?" -eq 124
MYSQL_PWD="$DB_PASS" mysql --protocol=TCP   -h "$DB_HOST" -P "$DB_PORT" -u "$DB_USER" "$DB_NAME"   -e "SELECT id, status, attempts FROM jobs WHERE id = $unsupported_id"

transient_id=$(
  MYSQL_PWD="$DB_PASS" mysql --protocol=TCP     -h "$DB_HOST" -P "$DB_PORT" -u "$DB_USER" "$DB_NAME" -Nse "
      INSERT INTO jobs (
        queue_name, job_type, payload, idempotency_key, max_attempts, available_at
      ) VALUES (
        'reports',
        'report.generate',
        JSON_OBJECT(
          'tenant_id', 'fixture-transient',
          'report_month', '2042-05',
          'test_transient_until_attempt', 2
        ),
        SHA2(CONCAT('fixture:transient:', UUID()), 256),
        4,
        UTC_TIMESTAMP(6)
      );
      SELECT LAST_INSERT_ID();
    "
)
timeout --signal=TERM 30s env WORKER_ENABLE_TEST_FIXTURES=1   php bin/worker.php 2> /tmp/report-worker-transient.log   || test "$?" -eq 124
MYSQL_PWD="$DB_PASS" mysql --protocol=TCP   -h "$DB_HOST" -P "$DB_PORT" -u "$DB_USER" "$DB_NAME"   -e "SELECT id, status, attempts FROM jobs WHERE id = $transient_id"

grep -q '"event":"job_failed"' /tmp/report-worker-transient.log
grep -q '"event":"job_completed"' /tmp/report-worker-transient.log

Nepodržani pomoćni scenarij mora biti dead nakon jednog pokušaja. Prolazni pomoćni scenarij mora doseći done pri trećem pokušaju, nakon dva sintetička neuspjeha i ograničenog eksponencijalnog čekanja s varijacijom.

Implementirajte sa systemdom

Instalirajte pregledane datoteke pod /opt/report-worker, u vlasništvu roota i čitljive namjenskom računu report-worker. Spremite dodjele varijabli okruženja u /etc/report-worker.env, u vlasništvu roota s načinom rada 0600. To su operacije administratora poslužitelja, a ne naredbe spremnika.

Izradite /etc/systemd/system/report-worker.service pomoću sudoedit i upotrijebite ovu točnu jedinicu:

[Unit]
Description=PHP report queue worker
After=network-online.target
Wants=network-online.target

[Service]
Type=simple
User=report-worker
Group=report-worker
WorkingDirectory=/opt/report-worker
EnvironmentFile=/etc/report-worker.env
ExecStart=/usr/bin/php /opt/report-worker/bin/worker.php
Restart=on-failure
RestartSec=3
KillSignal=SIGTERM
TimeoutStopSec=75
NoNewPrivileges=true
PrivateTmp=true
PrivateDevices=true
ProtectSystem=strict
ProtectHome=true
ProtectKernelTunables=true
ProtectKernelModules=true
ProtectControlGroups=true
LockPersonality=true
RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6
UMask=0077

[Install]
WantedBy=multi-user.target

Provjerite i omogućite točnu datoteku jedinice:

sudo systemd-analyze verify /etc/systemd/system/report-worker.service
sudo systemctl daemon-reload
sudo systemctl enable --now report-worker.service
sudo systemctl status report-worker.service
sudo journalctl -u report-worker.service --since today

Za dodatne instance upotrijebite systemd predložak. Postupno povećavajte broj: radnici troše veze s bazom podataka i nizvodni kapacitet čak i kada je sukob oko zaključavanja redaka nizak. Proračun za zaustavljanje od 75 sekundi premašuje najam, dok vremenska ograničenja veze, mrežnog čitanja i zaključavanja ostaju ispod oba.

Praćenje, sigurnost i performanse

Radnik emitira strukturirani JSON koji sadrži ID-jeve zadataka, pokušaje, proteklo vrijeme, brojeve oporavaka, gubitke najma, oslobađanja pri gašenju i događaje životnog ciklusa. Upozoravajte na rast mrtvih zadataka, starost najstarijeg spremnog zadatka, ponovljeni oporavak, stopu ponovnih pokušaja, latenciju obrade, neuspjehe oslobađanja i fatalne pogreške baze podataka.

Vjerodajnice i osjetljiva polja sadržaja držite izvan dnevnika i last_error. Provjeravajte sadržaje pri potrošnji čak i kada su proizvođači pouzdani. Za udaljeni MySQL zahtijevajte provjeru certifikata, ograničite port 3306 na ovlaštene adrese aplikacija i izradite račun ograničen na tu mrežu umjesto korištenja javnog zamjenskog hosta. Sam radnik ne treba dolazno pravilo vatrozida.

Mjerite planove upita za preuzimanje kako tablica raste. Dovršene zadatke zadržavajte samo onoliko dugo koliko zahtijevaju operativni ili revizijski zahtjevi te ih arhivirajte zasebnim pregledanim postupkom. Spori mrežni pozivi, generiranje izvješća i rad s datotečnim sustavom pripadaju izvan transakcija rezervacije i trebaju vlastite rokove kraće od najma.

Uobičajeni produkcijski neuspjesi

  • Radnici djeluju serijalizirano: obrada ostaje unutar transakcije preuzimanja ili nedostaje indeks za preuzimanje.
  • Zadaci se preklapaju nakon isteka: obrada traje dulje od najma. Ograničite vrijeme operacije, promišljeno prilagodite najam ili implementirajte obnovu ograđenu tokenom.
  • Upiti i dalje vise: konfiguriran je samo PDO::ATTR_TIMEOUT. Ograničenja veze, mrežnog čitanja i čekanja zaključavanja zasebne su kontrole; PDO MySQL nema univerzalno vremensko ograničenje upita.
  • Implementacije troše pokušaje: nedostaje provjera zaustavljanja nakon preuzimanja ili oslobađanje ne vraća neiskorišteni pokušaj.
  • Zastarjeli radnik potvrđuje: dovršavanje ne provjerava i token i neistekli rok pod zaključavanjem retka.
  • Pojavljuju se udaljeni duplikati: idempotentnost postoji samo u MySQL-u, a ne na granici vanjskog učinka.

Završni kontrolni popis za provjeru

  • Dvostruko stavljanje u red vraća jedan trajni ID zadatka.
  • Istodobni radnici preuzimaju različite prihvatljive retke bez dugih čekanja.
  • Rezervacija se potvrđuje prije početka poslovne obrade.
  • Ograničenja veze, mrežnog čitanja i čekanja zaključavanja kraća su od najma, a transakcije preuzimanja ostaju kratke i indeksirane.
  • Signal uočen tijekom claim() oslobađa samo odgovarajući token i ne pokreće obrađivač.
  • Istekli najmovi vraćaju se u ready dok preostaju pokušaji.
  • Iscrpljeni i trajni neuspjesi prelaze u dead.
  • Istekli ili zastarjeli token ne može dovršiti ni ponovno zakazati zadatak.
  • Tablica poslovnih učinaka odbacuje duplicirane idempotentne ključeve.
  • Odgode ponovnih pokušaja i najveći broj pokušaja konačni su.
  • Dnevnici otkrivaju latenciju, ponovne pokušaje, oporavak, gubitak najma, oslobađanje i gašenje.
  • Identitet tijekom izvođenja nema nepotrebne ovlasti za bazu podataka, datotečni sustav ni mrežu.

Trajni redak reda tek je početak otpornog radnika. Važan inženjering odvija se na granicama: kratko rezervirajte, izvršavajte bez zaključavanja, ogradite svako završno pisanje, oslobodite posao kada gašenje pobijedi u utrci i učinite stvarni učinak idempotentnim. Kad su ta pravila izričita, rušenja i ponovljena isporuka prestaju biti iznenađenja i postaju uobičajena stanja koja je sustav izgrađen podnijeti.

Portret autora bloga

Mihajlo

Ja sam Mihajlo — programer vođen znatiželjom, disciplinom i stalnom željom da stvorim nešto smisleno. Dijelim uvide, tutorijale i besplatne usluge kako bih pomogao drugima da pojednostave svoj rad i rastu u svijetu softvera i umjetne inteligencije koji se neprestano razvija.

Pretraži

Kategorije

Nedavni postovi

Product Ownership Beyond Feature Requests: Guiding Development with Intent
Napunite svoj fokus: održive navike za iscrpljenog radnika znanja
Kaldrmisane ulice i kreativne iskre: otkrivanje skrivenih čari Europe
I'Music – The Free Music Player That Finally Puts Music First
ApiQL: Izgradite REST API iz MySQL-a za nekoliko minuta