Consumer на Go: воркеры, ack/nack и graceful shutdown

// Запускаем 3 воркера
for i := 0; i < 3; i++ {
    go func(id int) {
        consumer := NewConsumer(ch, "progress-worker")
        consumer.Start(ctx, func(ctx context.Context, body []byte) error {
            log.Printf("worker %d processing message", id)
            return processMessage(body)
        })
    }(i)
}

<?php
declare(strict_types=1);

// В мире PHP «несколько consumer'ов на одну очередь» - это запуск нескольких процессов:
//   supervisor / systemd / docker compose --scale worker=3
// Внутри одного процесса используем один Consumer и пускаем его в while-цикле.

// Пример supervisord.conf:
//   [program:progress-worker]
//   command=php bin/console messenger:consume async --limit=1000 --time-limit=3600
//   numprocs=3
//   process_name=%(program_name)s_%(process_num)02d
//   autostart=true
//   autorestart=true

// А внутри одного воркера обработка остаётся обычным callback'ом:
final class WorkerEntrypoint
{
    public function __construct(
        private readonly Consumer $consumer,
        private readonly MessageProcessor $processor,
    ) {}

    public function run(): void
    {
        $this->consumer->start(function (string $body): void {
            $this->processor->process($body);
        });
    }
}

Распределение сообщений делает RabbitMQ - prefetch_count + равноценные consumer'ы дают round-robin между процессами.

func (c *Consumer) StartWithShutdown(ctx context.Context, handler func(context.Context, []byte) error) error {
    msgs, err := c.ch.Consume(c.queue, "", false, false, false, false, nil)
    if err != nil {
        return err
    }

    var wg sync.WaitGroup

    for {
        select {
        case <-ctx.Done():
            // Перестаём принимать новые сообщения
            c.ch.Cancel("", false)
            // Ждём завершения текущих
            wg.Wait()
            return nil

        case msg, ok := <-msgs:
            if !ok {
                wg.Wait()
                return nil
            }
            wg.Add(1)
            go func(m amqp.Delivery) {
                defer wg.Done()
                if err := handler(ctx, m.Body); err != nil {
                    m.Nack(false, true) // вернём в очередь
                    return
                }
                m.Ack(false)
            }(msg)
        }
    }
}

<?php
declare(strict_types=1);

use PhpAmqpLib\Channel\AMQPChannel;
use PhpAmqpLib\Message\AMQPMessage;

final class GracefulConsumer
{
    private bool $stopping = false;

    public function __construct(
        private readonly AMQPChannel $ch,
        private readonly string $queue,
    ) {
        // SIGTERM от docker / k8s, SIGINT от Ctrl+C
        pcntl_signal(SIGTERM, fn() => $this->stopping = true);
        pcntl_signal(SIGINT,  fn() => $this->stopping = true);
    }

    /** @param callable(string): void $handler */
    public function start(callable $handler): void
    {
        $this->ch->basic_consume(
            queue: $this->queue,
            no_ack: false,
            callback: function (AMQPMessage $msg) use ($handler): void {
                try {
                    $handler($msg->getBody());
                    $msg->ack();
                } catch (Throwable) {
                    $msg->nack(requeue: true); // вернём в очередь
                }
            },
        );

        while ($this->ch->is_consuming()) {
            pcntl_signal_dispatch();
            if ($this->stopping) {
                // Перестаём принимать новые сообщения; текущие уже доедут.
                $this->ch->basic_cancel('', noWait: false);
                break;
            }
            $this->ch->wait(timeout: 1.0);
        }
    }
}

В Symfony Messenger graceful shutdown настроен из коробки: воркер ловит SIGTERM сам и завершает текущее сообщение перед выходом.

func main() {
    ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
    defer cancel()

    conn, _ := amqp.Dial("amqp://guest:guest@localhost:5672/")
    defer conn.Close()

    ch, _ := conn.Channel()
    ch.Qos(10, 0, false)

    consumer := NewConsumer(ch, "progress-worker")

    log.Println("consumer started, waiting for messages...")
    if err := consumer.StartWithShutdown(ctx, handleMessage); err != nil {
        log.Fatalf("consumer error: %v", err)
    }
    log.Println("consumer stopped gracefully")
}

<?php
declare(strict_types=1);

// bin/consumer.php - точка входа worker-процесса.

use PhpAmqpLib\Connection\AMQPStreamConnection;

require __DIR__ . '/../vendor/autoload.php';

$conn = new AMQPStreamConnection('rabbitmq', 5672, 'guest', 'guest');
$ch = $conn->channel();
$ch->basic_qos(prefetch_size: 0, prefetch_count: 10, a_global: false);

$consumer = new GracefulConsumer($ch, 'progress-worker');

echo 'consumer started, waiting for messages...', "\n";
$consumer->start(handleMessage(...));
echo 'consumer stopped gracefully', "\n";

$ch->close();
$conn->close();

В Symfony - то же самое, но проще: php bin/console messenger:consume async --limit=1000. Внутри стоит --time-limit, ack/nack, graceful - всё уже из коробки.

func (c *Consumer) RunWithReconnect(ctx context.Context, dialURL string, handler func(context.Context, []byte) error) {
    for {
        select {
        case <-ctx.Done():
            return
        default:
        }

        conn, err := amqp.Dial(dialURL)
        if err != nil {
            log.Printf("connect failed: %v, retrying in 5s", err)
            time.Sleep(5 * time.Second)
            continue
        }

        ch, err := conn.Channel()
        if err != nil {
            conn.Close()
            continue
        }

        log.Println("connected to RabbitMQ")
        err = c.startConsuming(ctx, ch, handler)
        log.Printf("consumer stopped: %v, reconnecting...", err)

        ch.Close()
        conn.Close()
    }
}

<?php
declare(strict_types=1);

use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Exception\AMQPConnectionClosedException;
use PhpAmqpLib\Exception\AMQPIOException;
use Psr\Log\LoggerInterface;

final class ReconnectingConsumer
{
    public function __construct(
        private readonly LoggerInterface $logger,
        private readonly string $host,
        private readonly int $port,
        private readonly string $user,
        private readonly string $pass,
        private readonly string $queue,
    ) {}

    /** @param callable(string): void $handler */
    public function run(callable $handler): void
    {
        while (true) {
            try {
                $conn = new AMQPStreamConnection($this->host, $this->port, $this->user, $this->pass);
                $ch = $conn->channel();
                $this->logger->info('connected to RabbitMQ');

                $consumer = new Consumer($ch, $this->queue, $this->logger);
                $consumer->start($handler);
            } catch (AMQPConnectionClosedException | AMQPIOException $e) {
                $this->logger->warning('connection lost: {err}, retrying in 5s', ['err' => $e->getMessage()]);
                sleep(5);
                continue;
            } finally {
                if (isset($ch))   { $ch->close(); }
                if (isset($conn)) { $conn->close(); }
            }
        }
    }
}

Symfony Messenger воркер падает при потере соединения и поднимается заново supervisor'ом - так проще и надёжнее, чем переподключение в одном процессе.