Producer на Go: публикуем события в RabbitMQ

Producer (publisher) отправляет события в exchange. Его задача - надёжно доставить сообщение брокеру. Бизнес-правила остаются в домене, publisher - чистый транспорт (см. также event-driven pub/sub).

Структура события

Хорошее событие содержит всё, что нужно потребителю, и не заставляет его ходить в другие сервисы:

type LessonCompletedEvent struct {
    EventID     string    `json:"event_id"`
    Type        string    `json:"type"`
    UserID      int64     `json:"user_id"`
    LessonID    int64     `json:"lesson_id"`
    TrackID     int64     `json:"track_id"`
    CompletedAt time.Time `json:"completed_at"`
    Version     int       `json:"version"`
}

<?php
declare(strict_types=1);

final readonly class LessonCompletedEvent
{
    public function __construct(
        public string $eventId,
        public string $type,
        public int $userId,
        public int $lessonId,
        public int $trackId,
        public DateTimeImmutable $completedAt,
        public int $version,
    ) {}
}

Поле          Зачем
────────      ──────────────────────────────────────
event_id      Уникальный ID для идемпотентности и трейсинга
type          Тип события - consumer решает, обрабатывать или нет
version       Версия схемы - для обратной совместимости
*_id поля     Данные события - consumer не должен ходить в БД

UUID v4 - надёжный выбор. Вариант с `time.Now().UnixNano()` подходит для прототипов, но в продакшене при высокой нагрузке может дать коллизии. Используй `github.com/google/uuid`.

Простой Publisher

import amqp "github.com/rabbitmq/amqp091-go"

type RabbitPublisher struct {
    ch       *amqp.Channel
    exchange string
}

func NewRabbitPublisher(ch *amqp.Channel, exchange string) *RabbitPublisher {
    return &RabbitPublisher{ch: ch, exchange: exchange}
}

func (p *RabbitPublisher) Publish(ctx context.Context, routingKey string, event any) error {
    body, err := json.Marshal(event)
    if err != nil {
        return fmt.Errorf("marshal event: %w", err)
    }

    return p.ch.PublishWithContext(ctx,
        p.exchange,  // exchange
        routingKey,  // routing key
        false,       // mandatory
        false,       // immediate
        amqp.Publishing{
            ContentType:  "application/json",
            DeliveryMode: amqp.Persistent,  // сообщение переживёт рестарт
            MessageId:    uuid.New().String(),
            Timestamp:    time.Now(),
            Body:         body,
        },
    )
}

<?php
declare(strict_types=1);

use PhpAmqpLib\Channel\AMQPChannel;
use PhpAmqpLib\Message\AMQPMessage;
use Symfony\Component\Uid\Uuid;

final class RabbitPublisher
{
    public function __construct(
        private readonly AMQPChannel $ch,
        private readonly string $exchange,
    ) {}

    public function publish(string $routingKey, object $event): void
    {
        $body = json_encode($event, JSON_THROW_ON_ERROR);

        $msg = new AMQPMessage($body, [
            'content_type'  => 'application/json',
            'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT, // переживёт рестарт
            'message_id'    => Uuid::v4()->toRfc4122(),
            'timestamp'     => time(),
        ]);

        $this->ch->basic_publish($msg, $this->exchange, $routingKey);
    }
}

В Symfony Messenger обычно не пишут publisher руками - MessageBusInterface::dispatch($event) + AmqpTransport сделают всё то же самое плюс middleware-стек.

Persistent Messages

По умолчанию сообщения хранятся в памяти. При рестарте RabbitMQ - потеряются.

DeliveryMode: 1 (Transient)    Быстро, но ненадёжно
DeliveryMode: 2 (Persistent)   Медленнее, но сообщение на диске

// Persistent + durable queue = сообщение переживёт рестарт
amqp.Publishing{
    DeliveryMode: amqp.Persistent,  // = 2
    // ...
}

Persistent гарантирует запись на диск на стороне брокера. Но между `Publish()` и записью на диск может произойти сбой. Для полной гарантии нужен Publisher Confirm.

Publisher Confirms

Publisher Confirm - механизм, когда RabbitMQ подтверждает, что сообщение принято и записано:

func NewConfirmedPublisher(ch *amqp.Channel, exchange string) (*RabbitPublisher, error) {
    // Включаем режим подтверждений
    if err := ch.Confirm(false); err != nil {
        return nil, fmt.Errorf("enable confirms: %w", err)
    }
    return &RabbitPublisher{ch: ch, exchange: exchange}, nil
}

func (p *RabbitPublisher) PublishWithConfirm(ctx context.Context, routingKey string, event any) error {
    body, err := json.Marshal(event)
    if err != nil {
        return fmt.Errorf("marshal: %w", err)
    }

    // Получаем канал подтверждений
    confirmation, err := p.ch.PublishWithDeferredConfirmWithContext(ctx,
        p.exchange,
        routingKey,
        false,
        false,
        amqp.Publishing{
            ContentType:  "application/json",
            DeliveryMode: amqp.Persistent,
            MessageId:    uuid.New().String(),
            Body:         body,
        },
    )
    if err != nil {
        return fmt.Errorf("publish: %w", err)
    }

    // Ждём подтверждения от брокера
    ok, err := confirmation.WaitContext(ctx)
    if err != nil {
        return fmt.Errorf("wait confirm: %w", err)
    }
    if !ok {
        return fmt.Errorf("message nacked by broker")
    }

    return nil
}

<?php
declare(strict_types=1);

use PhpAmqpLib\Channel\AMQPChannel;
use PhpAmqpLib\Message\AMQPMessage;
use Symfony\Component\Uid\Uuid;

final class ConfirmedRabbitPublisher
{
    private bool $lastAcked = false;

    public function __construct(
        private readonly AMQPChannel $ch,
        private readonly string $exchange,
    ) {
        // Включаем режим подтверждений
        $this->ch->confirm_select();
        $this->ch->set_ack_handler(function (): void { $this->lastAcked = true; });
        $this->ch->set_nack_handler(function (): void { $this->lastAcked = false; });
    }

    public function publish(string $routingKey, object $event): void
    {
        $body = json_encode($event, JSON_THROW_ON_ERROR);

        $msg = new AMQPMessage($body, [
            'content_type'  => 'application/json',
            'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT,
            'message_id'    => Uuid::v4()->toRfc4122(),
        ]);

        $this->lastAcked = false;
        $this->ch->basic_publish($msg, $this->exchange, $routingKey);

        // Ждём подтверждения от брокера
        $this->ch->wait_for_pending_acks(timeout: 5.0);

        if (!$this->lastAcked) {
            throw new RuntimeException('message nacked by broker');
        }
    }
}

Routing Keys: конвенции именования

Хорошие routing key'и - иерархические и предсказуемые:

Паттерн: <domain>.<action>

lesson.completed     ✓  Понятно: урок завершён
lesson.started       ✓  Понятно: урок начат
user.registered      ✓  Понятно: пользователь зарегистрировался

update              ✗  Что обновилось?
event_1             ✗  Бессмысленно
lesson-completed    ✗  Точка, не дефис (topic exchange парсит по точкам)

Сериализация: JSON vs Protobuf

JSON                                 Protobuf
─────────────────────                ─────────────────────
Читаем глазами                       Нужен .proto файл
Гибкий, не требует схемы             Строгая типизация
Медленнее сериализация               Быстрее в 5-10 раз
Больше размер                        Компактнее в 3-5 раз
Хорош для старта                     Хорош для высоких нагрузок

Для большинства проектов JSON - достаточно. Переходи на Protobuf, когда размер сообщений или скорость сериализации станут узким местом.

Интеграция с Use Case

Publisher вызывается из use case, но сам не содержит бизнес-логики:

type CompleteLessonUseCase struct {
    progressRepo ProgressRepo
    lessonRepo   LessonRepo
    publisher    EventPublisher
}

func (uc *CompleteLessonUseCase) Execute(ctx context.Context, userID, lessonID int64) error {
    lesson, err := uc.lessonRepo.GetByID(ctx, lessonID)
    if err != nil {
        return err
    }

    // Бизнес-правила - в use case
    if err := uc.progressRepo.MarkCompleted(ctx, userID, lessonID); err != nil {
        return err
    }

    // Publisher - чистый транспорт
    return uc.publisher.Publish(ctx, "lesson.completed", LessonCompletedEvent{
        EventID:     uuid.New().String(),
        Type:        "lesson.completed",
        UserID:      userID,
        LessonID:    lessonID,
        TrackID:     lesson.TrackID,
        CompletedAt: time.Now(),
        Version:     1,
    })
}

<?php
declare(strict_types=1);

use Symfony\Component\Uid\Uuid;

final class CompleteLessonUseCase
{
    public function __construct(
        private readonly ProgressRepository $progressRepo,
        private readonly LessonRepository $lessonRepo,
        private readonly EventPublisher $publisher,
    ) {}

    public function execute(int $userId, int $lessonId): void
    {
        $lesson = $this->lessonRepo->getById($lessonId)
            ?? throw new LessonNotFoundException($lessonId);

        // Бизнес-правила - в use case
        $this->progressRepo->markCompleted($userId, $lessonId);

        // Publisher - чистый транспорт
        $this->publisher->publish('lesson.completed', new LessonCompletedEvent(
            eventId: Uuid::v4()->toRfc4122(),
            type: 'lesson.completed',
            userId: $userId,
            lessonId: $lessonId,
            trackId: $lesson->trackId,
            completedAt: new DateTimeImmutable(),
            version: 1,
        ));
    }
}

Обработка ошибок публикации

Что делать, если Publish() вернул ошибку?

Стратегия         Когда использовать
────────────      ──────────────────────────────────
Retry             Сеть мигнула, канал закрылся - попробуй снова
Outbox            Нужна гарантия: если БД сохранена, событие не потеряется
Log + alert       Некритичное событие (аналитика) - залогируй и иди дальше

Подробнее про Outbox - в уроке outbox/inbox. Retry с экспоненциальным backoff:

func (p *RabbitPublisher) PublishWithRetry(ctx context.Context, key string, event any, maxRetries int) error {
    var lastErr error
    for attempt := 0; attempt <= maxRetries; attempt++ {
        if err := p.Publish(ctx, key, event); err != nil {
            lastErr = err
            backoff := time.Duration(1<<attempt) * 100 * time.Millisecond // 100ms, 200ms, 400ms...
            time.Sleep(backoff)
            continue
        }
        return nil
    }
    return fmt.Errorf("publish failed after %d retries: %w", maxRetries, lastErr)
}

<?php
declare(strict_types=1);

final class RetryingPublisher
{
    public function __construct(
        private readonly RabbitPublisher $inner,
    ) {}

    public function publishWithRetry(string $routingKey, object $event, int $maxRetries): void
    {
        $lastError = null;
        for ($attempt = 0; $attempt <= $maxRetries; $attempt++) {
            try {
                $this->inner->publish($routingKey, $event);
                return;
            } catch (Throwable $e) {
                $lastError = $e;
                $backoffMs = (1 << $attempt) * 100; // 100ms, 200ms, 400ms...
                usleep($backoffMs * 1000);
            }
        }
        throw new RuntimeException(
            "publish failed after $maxRetries retries: " . $lastError?->getMessage(),
            previous: $lastError,
        );
    }
}

В Symfony Messenger ретраи настраиваются декларативно - retry_strategy.max_retries + multiplier в messenger.yaml, и брокер сам кладёт сообщения на failed транспорт.

Мини-задание

Напиши RabbitPublisher с методом Publish(ctx, routingKey, event) используя amqp091-go
Добавь DeliveryMode: amqp.Persistent и ContentType: "application/json"
Опубликуй тестовое событие lesson.completed и проверь в Management UI, что оно попало в нужную очередь
Добавь Publisher Confirm и убедись, что брокер подтвердил приём