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

func SetupRabbitMQ(conn *amqp.Connection) (*amqp.Channel, error) {
    ch, err := conn.Channel()
    if err != nil {
        return nil, fmt.Errorf("open channel: %w", err)
    }

    // Создаём exchange
    err = ch.ExchangeDeclare(
        "events",  // name
        "topic",   // type
        true,      // durable (переживёт рестарт)
        false,     // auto-deleted
        false,     // internal
        false,     // no-wait
        nil,       // arguments
    )
    if err != nil {
        return nil, fmt.Errorf("declare exchange: %w", err)
    }

    // Создаём очереди
    queues := []struct {
        name       string
        routingKey string
    }{
        {"progress-worker", "lesson.*"},
        {"analytics-worker", "lesson.*"},
        {"email-worker", "lesson.completed"},
    }

    for _, q := range queues {
        queue, err := ch.QueueDeclare(
            q.name,
            true,  // durable
            false, // auto-delete
            false, // exclusive
            false, // no-wait
            nil,
        )
        if err != nil {
            return nil, fmt.Errorf("declare queue %s: %w", q.name, err)
        }

        err = ch.QueueBind(
            queue.Name,
            q.routingKey,
            "events",
            false,
            nil,
        )
        if err != nil {
            return nil, fmt.Errorf("bind queue %s: %w", q.name, err)
        }
    }

    return ch, nil
}

<?php
declare(strict_types=1);

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Channel\AMQPChannel;

final class RabbitMqTopology
{
    public function __construct(
        private readonly AMQPStreamConnection $conn,
    ) {}

    public function setup(): AMQPChannel
    {
        $ch = $this->conn->channel();

        // Создаём exchange
        $ch->exchange_declare(
            exchange: 'events',
            type: 'topic',
            passive: false,
            durable: true,   // переживёт рестарт
            auto_delete: false,
        );

        // Создаём очереди + биндинги
        $queues = [
            ['name' => 'progress-worker',  'routingKey' => 'lesson.*'],
            ['name' => 'analytics-worker', 'routingKey' => 'lesson.*'],
            ['name' => 'email-worker',     'routingKey' => 'lesson.completed'],
        ];

        foreach ($queues as $q) {
            $ch->queue_declare(
                queue: $q['name'],
                passive: false,
                durable: true,
                exclusive: false,
                auto_delete: false,
            );
            $ch->queue_bind(
                queue: $q['name'],
                exchange: 'events',
                routing_key: $q['routingKey'],
            );
        }

        return $ch;
    }
}

В Symfony Messenger вместо ручной декларации топологии используют AmqpTransport (config/packages/messenger.yaml) - он сам создаёт exchange и очереди при первой публикации.

func ConnectRabbitMQ(url string) (*amqp.Connection, error) {
    // url формат: amqp://guest:guest@localhost:5672/
    conn, err := amqp.Dial(url)
    if err != nil {
        return nil, fmt.Errorf("dial rabbitmq: %w", err)
    }
    return conn, nil
}

// В docker-compose зависимый сервис:
// services:
//   app:
//     depends_on:
//       rabbitmq:
//         condition: service_healthy
//     environment:
//       RABBITMQ_URL: amqp://guest:guest@rabbitmq:5672/

<?php
declare(strict_types=1);

use PhpAmqpLib\Connection\AMQPStreamConnection;

final class RabbitMqConnector
{
    public function connect(string $host, int $port, string $user, string $pass): AMQPStreamConnection
    {
        // dsn-эквивалент: amqp://guest:guest@rabbitmq:5672/
        return new AMQPStreamConnection($host, $port, $user, $pass);
    }
}

// В docker-compose зависимый сервис:
// services:
//   app:
//     depends_on:
//       rabbitmq:
//         condition: service_healthy
//     environment:
//       MESSENGER_TRANSPORT_DSN: amqp://guest:guest@rabbitmq:5672/%2f/events