請求はキューで — 失わず、二度処理せず

목표

RabbitMQ 로 보험 청구를 비동기로 넘긴다. 토폴로지(교환기·큐·DLX)를 선언하고, 발행 확인을 받는 발행자와 수동 ack·DLQ·재배달·prefetch·중복 거르기를 갖춘 소비자를 만든다.

왜 중요한가

큐는 "잃어버리지 않는다" 를 약속하지만 그 약속은 발행 확인과 수동 ack 를 했을 때만 성립하고, 그 대가로 같은 메시지가 두 번 온다. 거절할 메시지를 되돌리면 독 메시지가 되고, prefetch 가 없으면 느린 대상 앞에서 소비자가 전부 떠안는다. 비동기 연동 사고의 대부분이 이 다섯 가지에서 난다.

단계

1. bash /opt/lab/fixtures/eaimw/mq/mq-up.sh 로 브로커를 띄우고 rabbitmqctl -n rabbit@localhost status > /root/eaimw/mq/status.txt 로 상태를 남긴다. 메모리 경보 기준이 절대값 512MiB(0.5369 gb)인지 확인한다.
2. /root/eaimw/mq/topology.py: 교환기 eai.claim(direct, durable), eai.dlx(direct, durable), 큐 claim.in(durable, 인자 x-dead-letter-exchange=eai.dlx, x-dead-letter-routing-key=claim.dead), claim.dead(durable), 바인딩 eai.claim→claim.in(키 claim), eai.dlx→claim.dead(키 claim.dead). 두 번 돌려도 오류가 없어야 한다.
3. /root/eaimw/mq/publish.py <청구JSON파일>: JSON 을 그대로 본문으로, eai.claim 에 라우팅 키 claim 으로 발행한다. persistent(delivery_mode 2), message_id 는 JSON 의 guid, content_type 은 application/json. 발행 확인을 켜고 mandatory 로 보내, 라우팅 불가나 거절이면 0 이 아닌 코드로 끝난다.
4. /root/eaimw/mq/consumer.py --claim <청구시스템URL> --max-seconds <초>: claim.in 을 소비해 본문을 POST /v1/claims 로 넘기고, 201 이면 그때 ack 한다(수동 ack). 그 밖의 결과는 일단 되돌린다(nack, requeue). --max-seconds 가 지나면 끝난다.
5. 업무 거절(422, 서류 미비)은 되돌리지 않고 reject(requeue=False) 로 DLX 에 보낸다.
6. 일시 오류(503·연결 실패·시간 초과)는 한 번 되돌리고, 재배달된(redelivered) 메시지가 또 실패하면 DLX 로 보낸다.
7. --prefetch(기본 5)로 basic_qos 를 걸어, 느린 청구 시스템 앞에서도 ack 하지 않은 메시지가 5건을 넘지 않게 한다.
8. 처리 기록을 SQLite(EAI_MQ_DB, 기본 /root/eaimw/mq/processed.db)에 남겨, 이미 처리한 message_id 는 청구 시스템을 부르지 않고 ack 만 한다(소비자를 다시 띄워도 기억한다).

참고

• 모든 스크립트는 EAI_AMQP_URL(기본 amqp://guest:guest@127.0.0.1:5672/%2F)을 읽는다: pika.BlockingConnection(pika.URLParameters(URL)). 채점기는 임시 vhost 주소를 넘긴다. guest 계정은 기본적으로 루프백에서만 접속할 수 있다.
• 청구 시스템 픽스처: nohup python3 /opt/lab/fixtures/eaimw/partner.py claim > /root/eaimw/mq/claim.out 2>&1 & (9202). docs 가 빈 청구는 422, claimId 가 BUSY 로 시작하면 503, 통계는 /_stats 의 by_guid(claimId 별 호출 수).
• 큐 들여다보기: rabbitmqctl -n rabbit@localhost list_queues name messages messages_unacknowledged.
• pika: ch.confirm_delivery() 뒤 basic_publish(..., mandatory=True) 가 라우팅 불가면 pika.exceptions.UnroutableError. 소비는 for m, props, body in ch.consume("claim.in", inactivity_timeout=0.5): (쉬는 동안 m 이 None).
• 흔한 실수: auto_ack=True, 업무 거절을 requeue, persistent 없이 durable 큐만 만들기.

파드 안에서 브로커를 띄운다

mq-up.sh 로 RabbitMQ 를 띄우고 rabbitmqctl status 출력을 /root/eaimw/mq/status.txt 에 남긴다.

도우미가 노드 이름을 rabbit@localhost 로 띄우므로 rabbitmqctl 에 -n rabbit@localhost 를 줍니다. 출력의 Memory high watermark 줄을 보세요.

교환기·큐·DLX 를 선언한다

/root/eaimw/mq/topology.py 가 교환기 2·큐 2·바인딩 2 와 claim.in 의 DLX 인자를 선언한다(두 번 돌려도 안전).

선언은 멱등합니다 — 같은 속성으로 다시 선언하면 아무 일도 없고, 속성이 다르면 브로커가 406 PRECONDITION_FAILED 로 채널을 닫습니다. DLX 는 큐 인자 두 개입니다.

확인을 받는 발행자

/root/eaimw/mq/publish.py 이 persistent·message_id·content_type 을 실어 발행하고, 발행 확인과 mandatory 로 라우팅 불가를 알아낸다.

ch.confirm_delivery() 를 먼저 부르면 basic_publish 가 브로커의 확인을 기다립니다. mandatory=True 로 보내면 받을 큐가 없을 때 UnroutableError 가 납니다.

처리한 뒤에만 ack 한다

/root/eaimw/mq/consumer.py 가 claim.in 을 소비해 청구 시스템에 넘기고, 201 을 받은 뒤에만 ack 한다(그 밖은 되돌림).

ch.consume(queue, inactivity_timeout=0.5) 로 돌면 쉬는 동안 None 이 옵니다. --max-seconds 가 지나면 빠져나와 연결을 닫으세요 — ack 안 한 것은 브로커가 되돌립니다. auto_ack 는 쓰지 않습니다.

업무 거절은 DLQ 로

422(서류 미비)는 되돌리지 말고 reject(requeue=False) 로 DLX 에 보낸다.

되돌린 메시지는 곧바로 다시 옵니다. 백 번 줘도 결과가 같은 메시지를 되돌리면 소비자가 그것만 붙잡고 돕니다. requeue=False 면 큐에 걸린 DLX 로 갑니다.

일시 오류는 한 번 더, 그래도 안 되면 DLQ

503·연결 실패는 한 번 되돌리고, redelivered 인데 또 실패하면 DLX 로 보낸다.

method 프레임의 redelivered 가 True 면 이미 한 번 되돌려진 메시지입니다. 되돌리기를 무한히 허용하면 대상이 오래 죽어 있을 때 큐 전체가 그 메시지들만 돕니다.

prefetch 로 떠안는 양을 제한한다

--prefetch(기본 5)로 basic_qos 를 걸어 unacked 가 5건을 넘지 않게 한다.

ch.basic_qos(prefetch_count=N) 은 채널 단위로 ack 하지 않은 배달 수를 제한합니다. 소비를 시작하기 전에 부르세요.

같은 message_id 는 한 번만

처리한 message_id 를 SQLite(EAI_MQ_DB)에 남겨, 다시 오면 부르지 않고 ack 만 한다.

처리에 성공하면 기록하고 ack 합니다. 기록과 ack 사이에 죽어도 다음에 온 같은 메시지는 기록 덕분에 걸러집니다. 기록은 소비자를 다시 띄워도 남아야 합니다.