Урок 02

Архитектура async-систем

Слои async-приложения

4 слоя:

  • API-слой — принимает HTTP, отвечает мгновенно.
  • Domain-слой — бизнес-логика.
  • Async-слой — очереди, воркеры, WebSocket.
  • Data-слой — БД, Redis, файлы.
HTTP → Controller → Service → Event → Queue
                                        ↓
                                     Worker → DB
                                        ↓
                                   Event → Node.js → WS

Event-Driven Architecture

Событие — что-то произошло. Слушатель — реагирует. Задача — что делать в фоне.

Пример:

// 1. Событие
class OrderCreated
{
    public function __construct(
        public readonly Order $order,
    ) {}
}

// 2. Controller
public function store(Request $request)
{
    $order = Order::create($request->validated());

    OrderCreated::dispatch($order);

    return response()->json($order, 201);
}

// 3. Слушатель
class SendOrderConfirmation
{
    public function handle(OrderCreated $event): void
    {
        SendEmailJob::dispatch($event->order->user_id)
            ->onQueue('emails');
    }
}

Что даёт:

  • Controller не знает про email.
  • Легко добавить нового слушателя.
  • События — точка расширения.

Producer-Consumer

Producer:

ProcessOrderJob::dispatch($order)
    ->onQueue('orders');

Broker:

Redis list: orders → [job1, job2, job3]

Consumer:

php artisan queue:work redis --queue=orders --tries=3

Воркеров может быть много:

                     → Worker 1
[Redis: orders] →    → Worker 2
                     → Worker 3

Каждая задача — одному воркеру.

Очереди по назначению

Разные очереди — разная нагрузка:

SendEmailJob::dispatch($user)->onQueue('emails');      // I/O
GenerateReportJob::dispatch($order)->onQueue('reports'); // CPU
ProcessVideoJob::dispatch($video)->onQueue('video');    // очень CPU

Supervisor по каждой:

[program:worker-emails]
command=php artisan queue:work redis --queue=emails
numprocs=10

[program:worker-reports]
command=php artisan queue:work redis --queue=reports
numprocs=3

[program:worker-video]
command=php artisan queue:work redis --queue=video
numprocs=1

Что даёт: CPU-задача не блокирует I/O-очередь.

Приоритеты

В одной очереди:

php artisan queue:work --queue=critical,high,default,low

Что происходит:

  1. Сначала critical.
  2. Потом high.
  3. Потом default.
  4. Потом low.

⚠️ Проблема: если critical не пустая — low никогда не выполнится.

Решение: отдельные воркеры для разных приоритетов.

Идемпотентность

Проблема: задача может выполниться дважды (retry, дубликат).

Решение — idempotency key:

class SendWelcomeEmail implements ShouldQueue, ShouldBeUnique
{
    public $uniqueFor = 3600;

    public function __construct(
        public int $userId,
    ) {}

    public function uniqueId(): string
    {
        return 'welcome:' . $this->userId;
    }
}

Или проверка в БД:

if (EmailLog::where('user_id', $this->userId)
    ->where('type', 'welcome')
    ->exists()) {
    return;
}

Dead Letter Queue

Проблема: задача упала после всех попыток — никто не знает.

Решение: перенос в DLQ.

public function failed(\Throwable $e): void
{
    DeadLetterJob::dispatch([
        'job' => static::class,
        'data' => ['userId' => $this->userId],
        'error' => $e->getMessage(),
    ])->onQueue('dlq');
}

Что даёт: можно разобрать вручную.

Circuit Breaker

Проблема: внешний API лежит, задача падает 100 раз.

Решение: после N ошибок — перестать вызывать.

use Illuminate\Support\Facades\Cache;

if (Cache::get('circuit:broken:stripe')) {
    $this->release(300);
    return;
}

try {
    $stripe->charge($order);
    Cache::forget('circuit:failures:stripe');
} catch (\Throwable $e) {
    $failures = Cache::increment('circuit:failures:stripe');

    if ($failures >= 5) {
        Cache::put('circuit:broken:stripe', true, 300);
    }

    throw $e;
}

Что даёт: не долбить упавший сервис.

Backpressure

Проблема: задачи поступают быстрее, чем обрабатываются.

Решение:

  • Ограничить rate: RateLimited middleware.
  • Проверять размер очереди:
if (Queue::size('emails') > 10000) {
    return response()->json(['error' => 'Queue overloaded'], 429);
}
  • Автоскейлинг: больше воркеров при росте очереди.

Retry стратегии

Фиксированная:

public array $backoff = [60, 60, 60];

Экспоненциальная:

public array $backoff = [10, 30, 90, 270];

Правило: сначала быстро (временный сбой), потом редко (внешний сервис восстанавливается).

Наблюдаемость

Что должно быть:

  • Логи — каждая задача с job_id.
  • Метрики — размер очереди, throughput, failed rate.
  • Трейсы — путь одной задачи через сервисы.
  • Алерты — при росте failed rate.

Пример: обработка заказа

1. HTTP: POST /orders
2. Controller: Order::create()
3. Event: OrderCreated
4. Слушатели (асинхронно):
   - SendConfirmationEmail → Queue: emails
   - ReserveInventory → Queue: inventory
   - NotifyWarehouse → Queue: warehouse
5. Воркеры обрабатывают
6. Event: OrderProcessed
7. Node.js подписан на Redis → WebSocket → клиент

HTTP отвечает за 20 мс. Всё остальное — в фоне.

Практика

Нарисуй схему для своего проекта:

  1. Какие операции синхронные?
  2. Какие можно в очередь?
  3. Какие очереди нужны?
  4. Где нужны воркеры, а где — WebSocket?

Реализуй Event-Driven:

  1. Создай событие UserRegistered.
  2. Слушатель SendWelcomeEmail.
  3. Слушатель NotifySlack.

Итог

  • 4 слоя: API, Domain, Async, Data.
  • Event-Driven: событие → слушатель → задача.
  • Очереди по назначению — I/O ≠ CPU.
  • Приоритеты — отдельные воркеры.
  • Идемпотентность — ShouldBeUnique или проверка.
  • DLQ — для failed.
  • Circuit Breaker — не долбить упавших.
  • Backpressure — не переполнять.
  • Наблюдаемость — логи, метрики, трейсы.

Дальше — API Gateway.