Consumer на Go: воркеры, ack/nack и graceful shutdown
Consumer - воркер, который читает сообщения из очереди и обрабатывает их. Главное правило: ack после успешной обработки. Всё остальное - следствие этого правила. См. также гарантии доставки.
Базовый Consumer
import amqp "github.com/rabbitmq/amqp091-go"
type Consumer struct {
ch *amqp.Channel
queue string
}
func NewConsumer(ch *amqp.Channel, queue string) *Consumer {
return &Consumer{ch: ch, queue: queue}
}
func (c *Consumer) Start(ctx context.Context, handler func(ctx context.Context, body []byte) error) error {
msgs, err := c.ch.Consume(
c.queue,
"", // consumer tag (пусто = автогенерация)
false, // auto-ack: false! Мы подтверждаем вручную
false, // exclusive
false, // no-local
false, // no-wait
nil,
)
if err != nil {
return fmt.Errorf("consume: %w", err)
}
for {
select {
case <-ctx.Done():
return ctx.Err()
case msg, ok := <-msgs:
if !ok {
return fmt.Errorf("channel closed")
}
if err := handler(ctx, msg.Body); err != nil {
// Обработка не удалась - nack с requeue
msg.Nack(false, true)
log.Printf("nack message %s: %v", msg.MessageId, err)
continue
}
// Успешно обработали - ack
msg.Ack(false)
}
}
}
<?php
declare(strict_types=1);
use PhpAmqpLib\Channel\AMQPChannel;
use PhpAmqpLib\Message\AMQPMessage;
use Psr\Log\LoggerInterface;
final class Consumer
{
public function __construct(
private readonly AMQPChannel $ch,
private readonly string $queue,
private readonly LoggerInterface $logger,
) {}
/** @param callable(string $body): void $handler */
public function start(callable $handler): void
{
$this->ch->basic_consume(
queue: $this->queue,
consumer_tag: '', // пусто = автогенерация
no_local: false,
no_ack: false, // ack: вручную! не auto-ack
exclusive: false,
nowait: false,
callback: function (AMQPMessage $msg) use ($handler): void {
try {
$handler($msg->getBody());
$msg->ack(); // успешно - ack
} catch (Throwable $e) {
// Обработка не удалась - nack с requeue
$msg->nack(requeue: true);
$this->logger->error('nack message {id}: {err}', [
'id' => $msg->get('message_id') ?? '',
'err' => $e->getMessage(),
]);
}
},
);
while ($this->ch->is_consuming()) {
$this->ch->wait();
}
}
}
В Symfony Messenger consumer запускается командой bin/console messenger:consume async - все нюансы ack/nack делает встроенный worker.
Ack, Nack, Reject
Метод Что происходит Когда использовать
────── ────────────────────── ──────────────────────
msg.Ack(false) Сообщение удалено из очереди Обработка успешна
msg.Nack(false, Сообщение возвращено в очередь Временная ошибка
true) (requeue = true) (сеть, таймаут)
msg.Nack(false, Сообщение удалено или в DLQ Перманентная ошибка
false) (requeue = false) (невалидные данные)
msg.Reject(true) = Nack(false, true) для 1 msg Альтернатива Nack
msg.Reject(false) = Nack(false, false) для 1 msg Отправить в DLQ
Ключевое: requeue: true возвращает сообщение в начало очереди. Если ошибка постоянная - получится бесконечный цикл. Используй счётчик ретраев:
func (c *Consumer) handleWithRetry(msg amqp.Delivery, handler func([]byte) error, maxRetries int) {
// Считаем ретраи через x-death header
retryCount := getRetryCount(msg)
if err := handler(msg.Body); err != nil {
if retryCount >= maxRetries {
// Отдаём в DLQ
msg.Nack(false, false)
log.Printf("message %s sent to DLQ after %d retries", msg.MessageId, retryCount)
return
}
// Ещё можно попробовать
msg.Nack(false, true)
return
}
msg.Ack(false)
}
func getRetryCount(msg amqp.Delivery) int {
xDeath, ok := msg.Headers["x-death"]
if !ok {
return 0
}
deaths, ok := xDeath.([]interface{})
if !ok || len(deaths) == 0 {
return 0
}
death, ok := deaths[0].(amqp.Table)
if !ok {
return 0
}
count, _ := death["count"].(int64)
return int(count)
}
<?php
declare(strict_types=1);
use PhpAmqpLib\Message\AMQPMessage;
use Psr\Log\LoggerInterface;
final class RetryConsumer
{
public function __construct(
private readonly LoggerInterface $logger,
private readonly int $maxRetries,
) {}
/** @param callable(string $body): void $handler */
public function handle(AMQPMessage $msg, callable $handler): void
{
// Считаем ретраи через x-death header
$retryCount = $this->getRetryCount($msg);
try {
$handler($msg->getBody());
$msg->ack();
} catch (Throwable $e) {
if ($retryCount >= $this->maxRetries) {
// Отдаём в DLQ
$msg->nack(requeue: false);
$this->logger->warning('message {id} sent to DLQ after {count} retries', [
'id' => $msg->get('message_id') ?? '',
'count' => $retryCount,
]);
return;
}
// Ещё можно попробовать
$msg->nack(requeue: true);
}
}
private function getRetryCount(AMQPMessage $msg): int
{
$headers = $msg->get_properties()['application_headers'] ?? null;
if ($headers === null) {
return 0;
}
$xDeath = $headers->getNativeData()['x-death'] ?? null;
if (!is_array($xDeath) || $xDeath === []) {
return 0;
}
return (int) ($xDeath[0]['count'] ?? 0);
}
}
Prefetch (QoS)
Prefetch контролирует, сколько сообщений consumer получает до подтверждения:
// Обрабатываем по одному - безопасно, но медленно
ch.Qos(1, 0, false)
// Обрабатываем пачками по 10 - быстрее, но нужен запас памяти
ch.Qos(10, 0, false)
Prefetch Поведение Когда
───────── ──────────────────────── ──────────────────
1 Одно сообщение за раз Тяжёлая обработка (email, PDF)
10-50 Пачки сообщений Лёгкая обработка (запись в БД)
0 Без ограничений (опасно!) Никогда в продакшене
Несколько Consumer'ов (Competing Consumers)
Запусти несколько consumer'ов на одну очередь - RabbitMQ распределит сообщения между ними:
// Запускаем 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 между процессами.
Queue: progress-worker
├── Consumer 1: обрабатывает msg 1, 4, 7...
├── Consumer 2: обрабатывает msg 2, 5, 8...
└── Consumer 3: обрабатывает msg 3, 6, 9...
RabbitMQ распределяет round-robin, но с учётом prefetch: свободный consumer получает следующее сообщение.
Graceful Shutdown через context
При остановке сервиса нужно дообработать текущие сообщения, а не бросать их:
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 сам и завершает текущее сообщение перед выходом.
Типичный сценарий entrypoint:
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 - всё уже из коробки.
Reconnect при потере соединения
Соединение с RabbitMQ может оборваться. Consumer должен уметь переподключаться:
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'ом - так проще и надёжнее, чем переподключение в одном процессе.
Мини-задание
- Напиши consumer с ручным ack/nack (auto-ack: false)
- Установи
prefetch = 1и обработай 10 сообщений - Добавь graceful shutdown через
signal.NotifyContext - Специально сломай обработку (return error) и проверь, что сообщение вернулось в очередь