Создание событийно-ориентированных рабочих процессов
Координируйте сервисы с помощью событий и идемпотентности
«Создание событийно-ориентированных рабочих процессов» — бесплатный урок PHP Academy на CoddyKit. Это урок 4 из 4. Ты можешь прочитать весь урок бесплатно ниже — а потом практиковать его прямо в браузере с встроенным редактором кода и ИИ-репетитором 24/7. Это часть пути обучения PHP Academy, и твой прогресс синхронизируется между веб-версией и приложением CoddyKit. Курс PHP Academy содержит 4 уроков всего.
Рабочие процессы на основе событий
Одно сообщение — это просто. Рабочий процесс — «заказ оформлен → зарезервировать товар → списать деньги с карты → отправить → уведомить», охватывающий несколько сервисов, — показывает преимущества проектирования на основе событий и создаёт проблемы при наивной реализации.
В этом уроке рассматриваются хореография и оркестрация, шаблон исходящих сообщений для атомарной публикации, саги для распределённого отката и идемпотентность, которая связывает всё это воедино.
Хореография и оркестрация
Есть два способа координировать многошаговые процессы:
- Хореография — каждый сервис реагирует на события и создаёт собственные; центрального управляющего нет. Сервисы слабо связаны, но общий процесс неявен и его трудно отслеживать.
- Оркестрация — центральный координатор сообщает каждому сервису, что делать дальше. Процесс явен и наблюдаем, но оркестратор становится точкой связанности.
Практическое правило: используйте хореографию для простой рассылки, а оркестрацию — когда процесс состоит из множества упорядоченных шагов и требует видимого состояния.
События и команды
Называйте сообщения осознанно:
- Событие сообщает о факте, который уже произошёл:
OrderPlaced. Издателю не важно, кто его слушает. - Команда запрашивает будущее действие у конкретного обработчика:
ChargeCard.
События управляют хореографией, а команды — оркестрацией. Смешение терминов, когда «событие» в действительности ожидает одного обработчика, часто приводит к скрытой связанности.
<?php
final class OrderPlaced {
public function __construct(
public readonly string $orderId,
public readonly string $customerId,
public readonly int $amountCents,
public readonly string $occurredAt,
) {}
}
$e = new OrderPlaced('o-42', 'c-7', 1990, gmdate('c'));
echo json_encode($e), "\n";Проблема двойной записи
Классическая ошибка: обработчик изменяет базу данных и публикует сообщение как две отдельные операции. Если процесс завершится между ними, возникнет несогласованность: строка изменится, но событие не будет отправлено, или произойдёт обратное.
<?php
// BROKEN: not atomic. A crash between the two lines corrupts state.
function placeOrder(PDO $db, $broker, array $o): void {
$db->prepare('INSERT INTO orders ...')->execute($o);
// <-- crash here = row exists but no event ever published
$broker->publish('OrderPlaced', json_encode($o));
}Транзакционный шаблон исходящих сообщений
Решение — шаблон исходящих сообщений: в рамках той же транзакции БД, которая изменяет данные, добавьте событие в таблицу outbox. Отдельный процесс-ретранслятор читает неопубликованные строки и отправляет их брокеру. Одна атомарная фиксация — никакой двойной записи.
<?php
function placeOrder(PDO $db, array $o): void {
$db->beginTransaction();
$db->prepare('INSERT INTO orders (id, total) VALUES (?, ?)')
->execute([$o['id'], $o['total']]);
// Same transaction -> atomic with the business write
$db->prepare('INSERT INTO outbox (id, type, payload) VALUES (?, ?, ?)')
->execute([bin2hex(random_bytes(8)), 'OrderPlaced', json_encode($o)]);
$db->commit();
}Ретранслятор (производитель)
Обработчик опрашивает таблицу исходящих сообщений (или отслеживает журнал изменений БД через CDC), публикует каждую строку, а затем помечает её как отправленную. Ретранслятор может завершиться после публикации, но до установки отметки, поэтому он сам обеспечивает доставку как минимум один раз. Это нормально, поскольку потребители идемпотентны.
<?php
function relayOutbox(PDO $db, $broker): void {
$rows = $db->query(
'SELECT id, type, payload FROM outbox
WHERE published_at IS NULL ORDER BY created_at LIMIT 100'
)->fetchAll(PDO::FETCH_ASSOC);
foreach ($rows as $r) {
$broker->publish($r['type'], $r['payload'], messageId: $r['id']);
$db->prepare('UPDATE outbox SET published_at = now() WHERE id = ?')
->execute([$r['id']]);
}
}Идемпотентные потребители (снова)
Поскольку и ретранслятор, и брокер обеспечивают доставку как минимум один раз, нижестоящие обработчики будут получать дубликаты. Каждый потребитель записывает идентификатор обработанного сообщения и пропускает повторы — это тот же шлюз дедупликации, который Вы изучали ранее, теперь применяемый отдельно в каждом сервисе.
<?php
function onOrderPlaced(PDO $db, string $messageId, array $data): void {
$db->beginTransaction();
try {
$db->prepare('INSERT INTO inbox (message_id) VALUES (?)')
->execute([$messageId]); // unique index = dedup
} catch (PDOException $e) {
$db->rollBack();
return; // already handled this message
}
reserveStock($data['orderId']);
$db->commit();
}
function reserveStock(string $id): void {}Саги: распределённый откат
Нельзя открыть одну транзакцию ACID между сервисами. Сага моделирует длительный процесс как последовательность локальных транзакций, каждая из которых имеет компенсирующее действие, отменяющее её результат. Если сбой произошёл на шаге 3, выполните компенсирующие действия для шагов 2 и 1 в обратном порядке.
Пример: платёж не прошёл после резервирования товара → отправьте ReleaseStock для компенсации. Автоматического отката нет — для каждого шага необходимо спроектировать действие отмены.
Сага с оркестрацией
Оркестратор управляет сагой: при успехе переводит её к следующему шагу, а при сбое запускает компенсирующие действия. Сохраняйте состояние саги, чтобы после сбоя можно было продолжить её выполнение.
<?php
function handleStepResult(array $saga, string $step, bool $ok, $bus): array {
if ($ok) {
$next = ['reserveStock' => 'chargeCard', 'chargeCard' => 'ship'][$step] ?? null;
if ($next) { $bus->send($next, $saga['orderId']); $saga['state'] = $next; }
else { $saga['state'] = 'completed'; }
} else {
// Run compensations in reverse for whatever already succeeded
foreach (array_reverse($saga['done']) as $s) {
$bus->send('compensate.' . $s, $saga['orderId']);
}
$saga['state'] = 'compensating';
}
return $saga;
}Времена ожидания в длительных процессах
Шаг саги может просто никогда не ответить: сервис оплаты недоступен или человек не подтверждает операцию. Без времени ожидания сага зависнет навсегда, удерживая резервы. Сохраняйте крайний срок для каждого шага; планировщик должен искать просроченные саги и запускать путь обработки сбоя и компенсации.
<?php
function reapTimedOutSagas(PDO $db, $bus): void {
$rows = $db->query(
"SELECT order_id, state FROM sagas
WHERE state NOT IN ('completed','compensating')
AND deadline_at < now()"
)->fetchAll(PDO::FETCH_ASSOC);
foreach ($rows as $r) {
echo "Saga {$r['order_id']} timed out at step {$r['state']}\n";
$bus->send('saga.compensate', $r['order_id']); // trigger rollback
}
}Версионирование и наблюдаемость
Рабочие процессы существуют годами; события должны безопасно развиваться:
- Добавляйте
version(или схему) к каждому событию; потребители должны корректно обрабатывать неизвестные новые поля и никогда не предполагать наличие поля. - Предпочитайте аддитивные изменения; никогда не переопределяйте смысл существующего поля.
- Передавайте корреляционный идентификатор через каждое сообщение, чтобы отслеживать одну бизнес-транзакцию во всех сервисах по журналам и трассировкам.
Без корреляционных идентификаторов отладка процесса с хореографией, проходящего через пять сервисов, практически невозможна.
<?php
$envelope = [
'type' => 'OrderPlaced',
'version' => 2,
'correlationId' => $incoming['correlationId'] ?? bin2hex(random_bytes(8)),
'occurredAt' => gmdate('c'),
'data' => ['orderId' => 'o-42'],
];
echo json_encode($envelope, JSON_PRETTY_PRINT), "\n";Быстрая проверка
Как избежать проблемы двойной записи.
Итоги
Теперь Вы умеете проектировать надёжные событийно-ориентированные рабочие процессы:
- Выбирайте хореографию (события) или оркестрацию (команды) в зависимости от сложности процесса.
- Решайте проблему двойной записи с помощью транзакционного исходящего журнала и ретранслятора.
- Делайте каждого потребителя идемпотентным с помощью входящего журнала и ключа устранения дубликатов.
- Используйте саги с компенсирующими действиями для распределённого отката.
- Версионируйте события, добавляя новые данные, и передавайте в каждом сообщении корреляционный идентификатор для отслеживания.
Эти шаблоны превращают разрозненные сообщения в надёжные и наблюдаемые бизнес-процессы.
Часто задаваемые вопросы
Урок «Создание событийно-ориентированных рабочих процессов» бесплатный?
Да — полный текст урока «Создание событийно-ориентированных рабочих процессов» бесплатно доступен здесь в веб-версии. Чтобы практиковать его интерактивно (встроенный редактор кода и ИИ-репетитор 24/7) и разблокировать остальной курс PHP Academy, подпишись на CoddyKit PRO. Курс PHP Academy содержит 4 уроков всего.
Чему я научусь в уроке «Создание событийно-ориентированных рабочих процессов»?
Координируйте сервисы с помощью событий и идемпотентности Ты практикуешь PHP Academy с помощью реального кода, который запускаешь прямо в браузере, и ИИ-репетитор 24/7 отвечает на твои вопросы во время урока.
Нужен ли мне опыт, чтобы начать PHP Academy?
Предыдущий опыт не требуется. PHP Academy на CoddyKit структурирован для всех уровней — от новичков до продвинутых, поэтому ты можешь начать отсюда или с самого начала и учиться в своем темпе. Это урок 4 из 4.
Сколько времени занимает урок «Создание событийно-ориентированных рабочих процессов»?
Большинство уроков CoddyKit занимают около 5–10 минут. Каждый из них компактный и интерактивный, поэтому ты постоянно делаешь прогресс и продолжаешь с того же места в веб-версии и приложении.
Можно ли писать и запускать код в этом уроке PHP Academy?
Да. Каждый урок PHP Academy включает встроенный редактор кода, поэтому ты пишешь и запускаешь реальный код прямо в браузере и получаешь моментальную обратную связь от AI — локальная установка не требуется.
Все уроки этого курса
- Зачем нужны асинхронные сообщения
- Работа с RabbitMQ в PHP
- Apache Kafka с PHP
- Создание событийно-ориентированных рабочих процессов