PHP 기반 RabbitMQ 메시지 파이프라인 설계: 신뢰성 발행-구독 모델과 재시도 및 사기 처리 구현

의존성 관리 및 환경 구성

RabbitMQ와 PHP 간 통신을 위한 공식 권장 클라이언트 라이브러리인 php-amqplib을 기반으로 프로젝트를 구성합니다.composer를 통해 의존성을 선언하면 네임스페이스 자동 로딩과 버전 충돌을 방지할 수 있습니다.

{
  "require": {
    "php": "^8.0",
    "php-amqplib/php-amqplib": "^3.6"
  }
}

터미널에서 다음 명령어로 패키지를 설치하거나 기존 프로젝트에 동적으로 추가할 수 있습니다.

composer require php-amqplib/php-amqplib

아키텍처 및 동작 원칙

본 구현체는 단순한 메시지 전달을 넘어 업무 로직의 신뢰성을 보장하기 위해 라비트큐의 고급 특성을 조합하여 설계되었습니다. 핵심 흐름은 다음과 같습니다.

  • Topic Exchange와 Work Queue 병행: 발행자는 주제 기반 라우팅(#, *)을 통해 메시지를 전송하고, 소비자는 경쟁 구조(Work Queue)로 큐에서 메시지를 가져와 병렬 처리합니다.
  • TTL와 Dead Letter Exchange(DLX) 기반 재시도: 처리 실패 시 즉시 버리지 않고, 일시적 대기열(TTL)에 배치합니다. 시간 초과 후 DLX를 거쳐 메인 교환기로 유입되어 자동으로 재소비됩니다.
  • 고정 횟수 초과 시 Dead Letter Queue: 최대 재시도 임계값을 초과하면 최종 실패 큐로 이동시켜 운영 팀의 개입이나 수동 복구 작업을 가능하게 합니다.
  • DB 연동 IDEM-POTENCY: 메시지의 고유 ID를 관계형 데이터베이스에 저장하여 발행 성공 여부 및 소비 중복을 검증하고, confirm_select 모드를 통해 전송 자체의 손실을 막습니다.

핵심 구성 요소 구현

1. 채널 및 연결 관리 계층

공통된 연결 설정과 리소스 관리를 담당하는 추상 클래스입니다. 연결 객체와 채널 인스턴스를 싱글톤 패턴으로 관리하여 중복 생성을 피하고, 명시적 닫힘 호출 시 자원 누수를 방지합니다.

<?php
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Channel\AMQPChannel;

abstract class AmqpBaseConnector
{
    protected ?AMQPStreamConnection $connection = null;
    protected ?AMQPChannel $channel = null;

    protected array $brokerConfig;

    public function __construct(array $config)
    {
        $this->brokerConfig = $config;
        $this->establishConnection();
    }

    private function establishConnection(): void
    {
        try {
            $this->connection = new AMQPStreamConnection(
                host: $this->brokerConfig['host'],
                port: $this->brokerConfig['port'] ?? 5672,
                login: $this->brokerConfig['user'],
                password: $this->brokerConfig['pass'],
                vhost: $this->brokerConfig['vhost'] ?? '/'
            );
            $this->channel = $this->connection->channel();
            // 기본 작업량 조정: 한 번에 50건까지만 할당하여 메모리 과부하 방지
            $this->channel->basic_qos(qos_size: null, prefetch_count: 50, global: false);
        } catch (\Exception $e) {
            throw new \RuntimeException('Broker 연결 실패: ' . $e->getMessage(), 0, $e);
        }
    }

    public function getChannel(): ?AMQPChannel
    {
        return $this->channel;
    }

    public function release(): void
    {
        if ($this->channel) $this->channel->close();
        if ($this->connection) $this->connection->close();
    }

    public function __destruct()
    {
        $this->release();
    }
}

2. 메시지 컨텍스트 래퍼

원본 AMQPMessage 객체를 감싸 비즈니스 로직에서 필요한 메타데이터(라우팅 키, 재시도 카운터 등)를 쉽게 접근할 수 있도록 캡슐화합니다.

<?php
use PhpAmqpLib\Message\AMQPMessage;
use PhpAmqpLib\Wire\AMQPTable;

class EnvelopeContext
{
    private AMQPMessage $originalMsg;
    private string $routingKey;
    private array $metaHeaders;

    public function __construct(AMQPMessage $msg, string $rKey, array $headers = [])
    {
        $this->originalMsg = $msg;
        $this->routingKey = $rKey;
        $this->metaHeaders = $headers;
    }

    public function getBody(): string
    {
        return $this->originalMsg->getBody();
    }

    public function decodeJson(): ?array
    {
        $decoded = json_decode($this->getBody(), true);
        return is_array($decoded) ? $decoded : null;
    }

    public function getUniqueIdentifier(): ?string
    {
        return $this->decodeJson()['uuid'] ?? null;
    }

    public function extractRetryCount(): int
    {
        if (!$this->originalMsg->has('application_headers')) return 0;
        
        $headers = $this->originalMsg->get('application_headers')->getNativeData();
        $deathInfo = $headers['x-death'][0] ?? [];
        return (int)($deathInfo['count'] ?? 0);
    }

    public function getOriginalRoutingKey(): string
    {
        return $this->metaHeaders['x-original-routing-key'] ?? $this->routingKey;
    }

    public function wrapForRePublish(): AMQPMessage
    {
        $properties = $this->originalMsg->get_properties();
        $newHeaders = new AMQPTable([
            'x-original-routing-key' => $this->getOriginalRoutingKey()
        ]);
        $properties['application_headers'] = $newHeaders;

        return new AMQPMessage(
            body: $this->getBody(),
            properties: $properties,
            delivery_mode: AMQPMessage::DELIVERY_MODE_PERSISTENT
        );
    }
}

3. 발행(Producer) 및 확인 모드 적용

전송 전 고유 ID 생성, DB 영속성 확보, 확인 선택(confirm_select) 활성화, 그리고 전송 성공/실패 시 각각의 콜백 핸들러를 등록합니다.

<?php
class MessagePublisher extends AmqpBaseConnector
{
    private string $mainExchange;
    private string $targetQueue;
    
    // 실제 시스템에서는 이 부분을 DI 또는 설정 파일로 대체 권장
    private DbRepositoryInterface $dbRepo;

    public function __construct(array $config, DbRepositoryInterface $dbRepo, string $exchange, string $queue)
    {
        parent::__construct($config);
        $this->mainExchange = $exchange;
        $this->targetQueue = $queue;
        $this->dbRepo = $dbRepo;
    }

    public function publish(array $payload, int $priority = 5): bool
    {
        $uuid = bin2hex(random_bytes(16));
        $payload['uuid'] = $uuid;
        $rawPayload = json_encode($payload, JSON_UNESCAPED_UNICODE);

        // 1. DB에 전송 예정 레코드 저장 (중복 전송 방지 및 추적용)
        if (!$this->dbRepo->saveOutgoing($uuid, $rawPayload, $priority)) {
            error_log("발행 전 DB 저장 실패: {$uuid}");
            return false;
        }

        // 2. 확인 모드 활성화
        $this->channel->confirm_select();

        // 3. ACK/NACK 콜백 정의
        $this->channel->set_ack_handler(function(AMQPMessage $msg) use ($uuid) {
            $this->dbRepo->markPublished($uuid);
        });
        $this->channel->set_nack_handler(function(AMQPMessage $msg) {
            error_log("브로커가 메시지를 거부함: " . $msg->getBody());
        });

        // 4. 메시지 생성 및 발행
        $amqpMsg = new AMQPMessage(
            body: $rawPayload,
            properties: [
                'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT,
                'priority'      => $priority,
                'content_type'  => 'application/json',
                'message_id'    => $uuid
            ]
        );

        $this->channel->basic_publish(msg: $amqpMsg, exchange: $this->mainExchange, routing_key: $this->targetQueue);

        // 전송 확인 대기
        $this->channel->wait_for_pending_acks();
        return true;
    }
}

4. 구독(Consumer) 및 재시도/사기 처리

큐를 바인딩하고, 소비자 태그를 비워두어 여러 프로세스가 경쟁하도록 합니다. no_ack=false로 설정하여 수동 확인만 허용하며, 예외 발생 시 TLV와 DLX를 활용한 지능형 재시도 로직을 적용합니다.

<?php
class TaskSubscriber extends AmqpBaseConnector
{
    private string $queueName;
    private string $retryExchange;
    private string $failExchange;
    private int $maxRetries = 3;

    public function __construct(array $config, string $qName, string $retryExch, string $failExch)
    {
        parent::__construct($config);
        $this->queueName = $qName;
        $this->retryExchange = $retryExch;
        $this->failExchange = $failExch;
    }

    /**
     * 리소스 초기화 및 루프 시작
     */
    public function run(callable $businessLogic, ?callable $onShutdown = null): void
    {
        $this->declareExchangesAndQueues();

        $stopPolling = false;
        if ($onShutdown) {
            pcntl_async_signals(true);
            pcntl_signal(SIGTERM, fn() => $stopPolling = true);
        }

        $this->channel->basic_consume(
            queue: $this->queueName,
            consumer_tag: '',
            no_local: false,
            no_ack: false,
            exclusive: false,
            nowait: false,
            callback: function(AMQPMessage $msg) use ($businessLogic, &$stopPolling) {
                if ($stopPolling) return;

                $context = new EnvelopeContext($msg, $msg->get('routing_key'));
                
                // 1. ID 중복 검사
                $uuid = $context->getUniqueIdentifier();
                if (!$uuid || !$this->checkIdempotency($uuid)) {
                    $this->channel->basic_ack(delivery_tag: $msg->delivery_info['delivery_tag']);
                    return;
                }

                try {
                    // 2. 비즈니스 로직 실행
                    $businessLogic($context);
                    
                    // 3. 성공 플래그 업데이트 및 확인 신호
                    $this->dbRepo->markConsumed($uuid);
                    $this->channel->basic_ack(delivery_tag: $msg->delivery_info['delivery_tag']);
                } catch (\Throwable $ex) {
                    $retryCount = $context->extractRetryCount();
                    $this->handleFailure($context, $msg, $retryCount);
                }
            }
        );

        // 이벤트 루프
        while (count($this->channel->callbacks)) {
            try {
                $this->channel->wait(timeout: 2.0);
            } catch (\PhpAmqpLib\Exception\AMQPIOWaitException| \PhpAmqpLib\Exception\AMQPTimeoutException $e) {
                // 타임아웃은 정상적인 폴링 상태이므로 무시
            }
        }
    }

    private function handleFailure(EnvelopeContext $ctx, AMQPMessage $msg, int $currentRetry): void
    {
        if ($currentRetry >= $this->maxRetries) {
            // 최대 초과 시 실패 큐로 포워딩
            $failedMsg = $ctx->wrapForRePublish();
            $this->channel->basic_publish($failedMsg, $this->failExchange, $ctx->getOriginalRoutingKey());
        } else {
            // 재시도를 위한 지연 발송 (DLX 기능 활용)
            $retryMsg = $ctx->wrapForRePublish();
            $this->channel->basic_publish($retryMsg, $this->retryExchange, $ctx->getOriginalRoutingKey());
        }
        // 실패 시에도 확인 신호 제거하여 브로커가 다음 기회를 부여하거나 재전송됨
    }

    private function declareExchangesAndQueues(): void
    {
        // 교환기 설정
        foreach ([$this->retryExchange, $this->failExchange] as $ex) {
            $this->channel->exchange_declare($ex, 'topic', passive: false, durable: true, auto_delete: false);
        }

        // 메인 큐
        $this->channel->queue_declare(
            queue: $this->queueName,
            passive: false,
            durable: true,
            exclusive: false,
            auto_delete: false,
            nowait: false,
            arguments: ['x-max-priority' => 10]
        );
        $this->channel->queue_bind($this->queueName, $this->retryExchange, '#');
        $this->channel->queue_bind($this->queueName, $this->failExchange, '#');

        // 재시도 전용 큐 (TTL + DLX 설정)
        $retryQueue = "{$this->queueName}@retry";
        $this->channel->queue_declare(
            queue: $retryQueue,
            passive: false,
            durable: true,
            exclusive: false,
            auto_delete: false,
            nowait: false,
            arguments: [
                'x-dead-letter-exchange' => '', // 빈 문자열은 메인 큐로 복귀 의미
                'x-dead-letter-routing-key' => $this->queueName,
                'x-message-ttl' => 5000 // 5초 대기 후 재전송
            ]
        );
    }
    
    // 실제 구현 시에는 DbRepositoryInterface의 메서드 구현 필요
    private function checkIdempotency(string $uuid): bool { return true; }
}

데이터 일관성 보장 스키마

발행자와 소비자 양쪽에서 고유 식별자를 기준으로 상태를 추적합니다. InnoDB 엔진을 사용하고 복합 조건 검색을 최적화하기 위해 적절한 인덱스를 구성합니다.

CREATE TABLE `ampq_task_tracker` (
  `task_uuid` VARCHAR(40) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL,
  `payload_data` TEXT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NULL,
  `expected_priority` TINYINT UNSIGNED DEFAULT 5,
  `status_flag` TINYINT(1) UNSIGNED NOT NULL DEFAULT 0 COMMENT '0:출력대기, 1:전송완료, 2:소비완료',
  `retry_attempt` INT UNSIGNED DEFAULT 0,
  `created_at` INT UNSIGNED NOT NULL,
  `updated_at` INT UNSIGNED NOT NULL,
  PRIMARY KEY (`task_uuid`),
  KEY `idx_status_created` (`status_flag`,`created_at`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci ROW_FORMAT=DYNAMIC;

실행 흐름 요약

  1. 发布: 발행 측에서 UUID를 생성하고 DB에 status_flag=0로 기록한 뒤 confirm_select 모드로 메시지를 브로커에 보냅니다.
  2. 라우팅: Topic Exchange가 라우팅 키와 일치하는 규칙을 찾아 지정된 큐에 배정합니다.
  3. 경쟁 소비: 여러 Worker 프로세스가 basic_consume를 통해 큐에서 잠금을 잡고 메시지를 가져옵니다.
  4. 일관성 검증: 소비자가 들어온 순간 DB에서 해당 UUID의 status_flag를 체크합니다. 이미 2라면 건너뛰고, 1이면 처리 후 2로 변경합니다.
  5. 예외 및 재시도: 로직 실행 중 오류가 발생하면 DLX 기능을 가진 지연 큐로 메시지를 넘깁니다. TTL이 만료되면 브로커가 자동으로 원본 큐로 재배치합니다.
  6. 최종 실패 처리: 최대 횟수(예: 3회) 이상 실패하면 failExchange로 이동해 운영팀의 모니터링 도구 또는 대시보드에서 확인할 수 있게 됩니다.
  7. 확인 신호: 모든 단계에서 basic_ack 또는 basic_nack을 적절히 호출하여 브로커의 메시지 회수(재전송) 순환 문제를 차단합니다.

태그: RabbitMQ php-amqplib message-queue Idempotency dead-letter-exchange

9월 9일 06:01에 게시됨

인기 태그

©2026 괴물 클럽