LASSIC Media らしくメディア

監修・編集責任者：牛尾 昭昌（株式会社LASSIC 執行役員）

非同期の処理をつなぐためにRabbitMQを入れたものの、キューにメッセージが溜まり続ける、再起動したらメッセージが消えていた——。RabbitMQを使い始めた現場では、こうしたつまずきが起こりがちです。RabbitMQは、AMQP 0-9-1などのプロトコルでアプリケーションどうしをつなぐメッセージブローカーです。

本記事では、システムの開発と運用を担当する方に向けて、Exchange・Binding・Queueの関係、Exchangeの型の使い分け、Pythonのpikaでのコード、取りこぼさずに届けるためのackとpublisher confirm、つまずきやすい点を、公式ドキュメントをもとに整理します。

RabbitMQとは

RabbitMQは、メッセージを仲介するソフトウェア（メッセージブローカー）です。公式ドキュメントは、ブローカーの役割を、メッセージを送る側（publisher）から受け取り、処理する側（consumer）へ振り分けることだと説明しています。^*1 送信側・受信側・ブローカーは、それぞれ別のマシンに置けます。

間にキューを挟む利点は製品を問わない話なので「メッセージキューとは」に譲り、本記事ではRabbitMQに固有の部分に絞ります。いちばんの特徴は、送信側がキューへ直接入れるのではなく、間にExchangeという振り分け役を挟むことです。

公式のリリース情報では、2026年10月1日時点の最新の版は2026年9月16日公開の4.3.6で、4.3系のコミュニティサポートの終了日は2026年11月30日とされています。^*2 本記事の説明は、公式ドキュメントの4.3版に沿っています。

Exchange・Binding・Queueの関係

AMQP 0-9-1の考え方では、メッセージはまずExchangeへ送られます。ExchangeはBindingという規則に従ってQueueへメッセージの写しを配り、ブローカーはQueueを購読している受信側へメッセージを届けます。^*1

Bindingは、どのExchangeからどのQueueへ届けるかを決める規則です。Bindingにはrouting keyを付けられ、Exchangeはメッセージに付いたrouting keyと照らし合わせて届け先を選びます。送信側はQueueの名前を知らなくても、Exchangeの名前とrouting keyだけで送れるため、受信側を増やしたり外したりしても送信側のコードは変わりません。

どのBindingにも合わないメッセージは、送信側が付けた属性によって、捨てられるか送信側へ戻されます。戻してもらうにはmandatoryを付けます。また、同じ名前のQueueを違う属性で宣言し直すと、406（PRECONDITION_FAILED）のエラーになります。

Exchangeの4つの型の使い分け

Exchangeには、direct・fanout・topic・headersの4つの型があり、型によってBindingとの照らし合わせ方が変わります。

topic型では、routing keyをドットで区切った語の並びにします。長さは255バイトまでで、Binding側のパターンでは、*がちょうど1語に、#が0語以上に当たります。^*3 下は、障害の通知を受け持つQueueを、2つのパターンで結ぶ例です。

import pika

conn = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
ch = conn.channel()
ch.exchange_declare(exchange='events', exchange_type='topic', durable=True)
ch.queue_declare(queue='alerts', durable=True)
ch.queue_bind(exchange='events', queue='alerts', routing_key='*.critical')
ch.queue_bind(exchange='events', queue='alerts', routing_key='payment.#')
for key in ['inventory.critical', 'payment.api.error', 'inventory.info']:
    ch.basic_publish(exchange='events', routing_key=key, body=key.encode())
conn.close()

inventory.criticalは*.criticalに、payment.api.errorはpayment.#に合うのでalertsへ届き、inventory.infoはどちらにも合わないため捨てられます。なお、この環境ではRabbitMQのサーバーを動かせなかったため、本記事のコードは、pikaのメソッドに渡す引数の確認と、公式チュートリアルにある照合の例を使った振り分けの検算までにとどめています。

型は、まずdirect型で足りるかを考えます。種類ごとに届け先が決まっているならdirect型、受信側が欲しい種類だけを選んで受け取るならtopic型、全員へ同じものを配るならfanout型が向きます。

Pythonのpikaで送る・受け取る

公式チュートリアルが使っているPythonのクライアントpikaで書いてみます。まずは送信側で、direct型のExchangeとquorum queueを結び、persistentのメッセージを送ります。

import pika

conn = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
ch = conn.channel()
ch.exchange_declare(exchange='orders', exchange_type='direct', durable=True)
ch.queue_declare(queue='order.created', durable=True,
                 arguments={'x-queue-type': 'quorum'})
ch.queue_bind(exchange='orders', queue='order.created', routing_key='created')
ch.confirm_delivery()  # publisher confirm を有効にする
try:
    ch.basic_publish(exchange='orders', routing_key='created',
                     body=b'{"order_id": 1001}',
                     properties=pika.BasicProperties(
                         delivery_mode=pika.DeliveryMode.Persistent),
                     mandatory=True)
except (pika.exceptions.UnroutableError, pika.exceptions.NackError):
    print('届いていない。再送するか記録に残す')
conn.close()

confirm_deliveryを呼ぶと、そのチャネルでpublisher confirmが有効になります。pikaでは、mandatoryを付けたメッセージがどのQueueにも届かず戻されるとUnroutableErrorが、ブローカーが受け取りを断るとNackErrorが上がります。^*4

次は受信側です。1回に預かる未確認のメッセージの数を決め、処理が終わってからackを返します。

import pika

def process(body):
    print(body)  # ここに業務処理を書く（同じ注文が2回来ても困らない作りに）

def handle(ch, method, properties, body):
    try:
        process(body)
        ch.basic_ack(delivery_tag=method.delivery_tag)
    except Exception:
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)

conn = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
ch = conn.channel()
ch.basic_qos(prefetch_count=50)
ch.basic_consume(queue='order.created', on_message_callback=handle)
ch.start_consuming()

pikaでは、auto_ackを指定しなければ手動のackになります。^*5 失敗したメッセージをrequeue=Falseで返すと、捨てられるか、退避先のExchangeが設定してあればそちらへ回されます。requeue=Trueで戻すと、同じメッセージを何度も受け取り直す繰り返しになりやすいので注意します。退避先の考え方は「デッドレターキューとは」で扱っています。

ackとpublisher confirmで届ける

取りこぼしを防ぐ仕組みは2つあります。受信側が処理の終わりを知らせるack（確認応答）と、ブローカーが受け取ったことを送信側に知らせるpublisher confirmです。2つは別の区間を受け持つので、両方をそろえて初めて送信側から受信側までがつながります。

自動のackでは、ブローカーが送った時点で届いたものとして扱うため、処理の前に接続が切れるとメッセージは失われます。手動のackなら、ackを返していないメッセージは、チャネルや接続が閉じたときにQueueへ戻され、別の受信側に再配信されることもあります。受信側は、同じ注文を2回受け取っても二重に処理しない作りにしておきます。

送信側も、ソケットに書き込んだだけではブローカーに届いたとは言えません。AMQP 0-9-1の標準ではトランザクションしか方法がなく、公式ドキュメントはトランザクションが処理量を250分の1に落とすと説明しています。^*6 その代わりに用意されたのがpublisher confirmです。persistentのメッセージをdurableなQueueへ送った場合はディスクへ書いた後、quorum queueの場合は過半数の複製が受け入れた後に、basic.ackが返ります。

この返事は、負荷が続くと数百ミリ秒かかることがあります。上の送信側のコードは1件ごとに返事を待つため、件数の多い処理では、まとめて送ってから未確認の分を待つ書き方にします。

キューの種類とStreams

RabbitMQの置き場所には、quorum queue、classic queue、streamの3つがあります。

quorum queueは、受注のように1件の欠落がシステムの正しさに響く、長く使い続けるQueueを対象としており、一時的なQueueや遅延をできるだけ小さくしたい処理には向かないとされています。^*7 4.0からは、Queueへ戻された回数の上限（delivery-limit）の既定が20になりました。classic queueの複製（ミラーリング）は4.x系で取り除かれているため、3.x系から移すときは種類の選び直しが要ります。

streamは、読んでも消えず何度でも読み返せるログです。^*9 同じものを大勢に配る処理や、過去のメッセージを読み直したい処理で検討します。

つまずきやすい点

ackの返し忘れ　公式チュートリアルは、basic_ackの書き忘れをよくある間違いとし、影響は深刻だとしています。^*5 未確認のメッセージを解放できず、ブローカーのメモリーの使用量が増え続けます。rabbitmqctl list_queues name messages_ready messages_unacknowledgedで、未確認の数を確かめられます。ackの待ち時間の上限は既定で30分で、超えるとチャネルが閉じられます（4.3からはquorum queueだけ）。^*10

prefetchを決めていない　basic.qosのprefetch countは、未確認のまま預かれるメッセージの数です。0は上限なしで、大量に受け取ると受信側のメモリーが膨らみます。公式ドキュメントは、100〜300の範囲がふつうはよい処理量になり、1は処理量を大きく落とすと説明しています。^*6 決めないと、重い処理を抱えた受信側にも順番に配られ続け、負荷が偏ります。

durableとpersistentの片方だけ　再起動の後もメッセージを残すには、Queueをdurableにし、メッセージをpersistentで送る、の両方が要ります。durableなQueueでも、persistentでないメッセージは再起動のときに捨てられます。^*8 persistentでもディスクへ書く前のわずかな間は失われることがあるため、publisher confirmを組み合わせます。

キューの肥大化　受信側が追いつかないとQueueは溜まり続けるので、受信側を増やすか、TTL（有効期限）を付けるかを決めておきます。長さの上限を付けると、既定では古いメッセージから捨てるか退避させ、overflowをreject-publishにすると新しいものを断ります。未確認のメッセージは上限の数に含まれません。^*11

外部に委託するときに確認しておきたい点

委託するときは、3つの点を確かめておくと比べやすくなります。第一に、Exchange・Queue・Bindingの定義を誰がどこで管理するかです。Queueの長さの上限のような設定は、アプリケーションの引数で固定するより、後から変えられるポリシーで設定することを公式ドキュメントも勧めています。

第二に、届け方の約束です。手動のackとpublisher confirmを使うか、再配信で2回届いたときにどう扱うか、何回失敗したら退避させるかを設計書に書いてもらいます。

第三に、運用の手当てです。溜まっている件数、未確認の件数、受信側の数を監視して通知する仕組みがあるか、版のサポート期限と版上げの試験の進め方を確かめます。検証用の環境でブローカーをわざと再起動し、メッセージが残るかを確かめる試験があるかも見ておきます。

まとめ：RabbitMQで確かめておきたい3つの点

RabbitMQで確かめておきたい点は3つです。第一に、ExchangeがBindingの条件に合うQueueへ写しを配る関係を押さえ、届け先の決め方に合わせて型を選ぶこと。第二に、手動のackとpublisher confirmを使い、durableとpersistentをそろえ、受信側を再配信に備えた作りにすること。第三に、prefetchを決め、溜まっている件数と未確認の件数を監視することです。

よくある質問

出典

1. *1　参考：RabbitMQ Documentation「AMQP 0-9-1 Model Explained」（https://www.rabbitmq.com/tutorials/amqp-concepts）。出典：ブローカーの役割、Exchange・Queue・Bindingの関係、4つのExchangeの型と既定の名前、既定のExchange、ルーティングできないメッセージの扱い、406（PRECONDITION_FAILED）、push型とpull型を参照（2026年10月確認）
2. *2　参考：RabbitMQ「Release Information」（https://www.rabbitmq.com/release-information）。出典：4.3.6の公開日（2026年9月16日）と、4.3系のコミュニティサポートの終了日（2026年11月30日）を参照（2026年10月1日確認）（2026年10月確認）
3. *3　参考：RabbitMQ Tutorials「Topics（Python）」（https://www.rabbitmq.com/tutorials/tutorial-five-python）。出典：topic型のrouting keyの形（ドット区切り・255バイトまで）、*と#の意味、照合の例を参照（2026年10月確認）
4. *4　参考：Pika Documentation「BlockingConnection」（https://pika.readthedocs.io/en/stable/modules/adapters/blocking.html）。出典：BlockingChannelのbasic_publish（mandatory、UnroutableError、NackError）、confirm_delivery、basic_consume、basic_qosを参照（pika 1.4.4）（2026年10月確認）
5. *5　参考：RabbitMQ Tutorials「Work Queues（Python）」（https://www.rabbitmq.com/tutorials/tutorial-two-python）。出典：手動のackが既定であること、basic_ackの書き忘れとその影響、messages_unacknowledgedの確かめ方、durableとpersistentの両方が要ること、Queueが溜まったときの対処を参照（2026年10月確認）
6. *6　参考：RabbitMQ Documentation「Consumer Acknowledgements and Publisher Confirms」（https://www.rabbitmq.com/docs/confirms）。出典：自動と手動のack、再配信と冪等性、publisher confirmの仕組み、トランザクションとの処理量の違い、basic.ackの時期と遅延、prefetchの値の目安を参照（2026年10月確認）
7. *7　参考：RabbitMQ Documentation「Quorum Queues」（https://www.rabbitmq.com/docs/quorum-queues）。出典：quorum queueの仕組み（Raft、過半数）、向く使い方と向かない使い方、delivery-limitの既定（4.0から20）を参照（2026年10月確認）
8. *8　参考：RabbitMQ Documentation「Queues」（https://www.rabbitmq.com/docs/queues）。出典：durableとtransientのQueue、再起動のときにpersistentでないメッセージが捨てられること、classic queueの複製が4.x系で取り除かれたことを参照（2026年10月確認）
9. *9　参考：RabbitMQ Documentation「Streams and Superstreams」（https://www.rabbitmq.com/docs/streams）。出典：streamが追記だけのログで読み返せること、Queueを補うものであること、使いどころを参照（2026年10月確認）
10. *10　参考：RabbitMQ Documentation「Consumers」（https://www.rabbitmq.com/docs/consumers）。出典：ackの待ち時間の上限（既定30分、超えるとチャネルを閉じる、4.3からquorum queueだけ）を参照（2026年10月確認）
11. *11　参考：RabbitMQ Documentation「Queue Length Limit」（https://www.rabbitmq.com/docs/maxlength）。出典：Queueの長さの上限に達したときの既定の動き、overflowのreject-publish、未確認のメッセージを数えないこと、ポリシーでの設定を勧めていることを参照（2026年10月確認）

Tweet