はじめに
Azure Service Bus のセッション対応キューを使っていると、セッション内のメッセージがなくなったときにすぐにセッションを離脱してNEXT_AVAILABLE_SESSIONで新しいセッションを掴みにいきたいけど、新しいセッションが見つからないときには待機時間を長くしてポーリングのし過ぎを防ぎたい、という状況に直面することがあります。この記事では、そんなときの工夫について紹介します。
TL;DR
ServiceBusClientのmax_wait_timeはget_queue_receiverを呼んだときにだけ設定できるわけじゃなく、receive_messagesでも上書きできるので、セッションの取得とメッセージの受信で異なる待機時間を設定することが可能です。
セッションって何？
Service Bus のセッションは、同じ session_id を持つメッセージをグループ化して、順序を保って処理するための機能です。FIFO（先入れ先出し）や、関連するメッセージの一括処理、排他処理に便利です。
Azure Service Bus のメッセージ セッション - Microsoft Learn
状態遷移図で見るセッション管理
セッションを使用しない場合
Microsoft Learnで提案されているようなの利用法の場合、以下のような状態遷移が発生します：
ここで、NEXT_AVAILABLE_SESSIONを利用する場合はOperationTimeoutErrorの取得により、特定のセッションにはメッセージがない、という情報を取得し、次のセッションへ移行するという動作が行われますが、max_wait_timeが30秒などの長い場合にはセッションへの移行が遅れ、0.5秒などの短い場合にはポーリングが起きすぎるという問題が発生し得ます。
具体的な解決策
以下のように、 セッション管理を最適化できます。
これを解決するために、セッションがない初期状態では長めの待機時間をmax_wait_timeに設定し、セッションがある状態では短めの待機時間でメッセージを受信することで、効率的なセッション管理が可能になります。
import asyncio
from azure.servicebus.aio import ServiceBusClient, NEXT_AVAILABLE_SESSION
async def process_messages():
servicebus_client = ServiceBusClient.from_connection_string("YourConnectionString")
async with servicebus_client:
while True:
# receiverに短めの待機時間を設定
async with servicebus_client.get_queue_receiver(
queue_name="your_queue_name",
session_id=NEXT_AVAILABLE_SESSION,
max_wait_time=0.5
) as receiver:
# セッション内のメッセージを受信する際は、長めの待機時間を別途設定
try:
messages = await receiver.receive_messages(max_wait_time=30)
except OperationTimeoutError:
continue
if not messages:
# メッセージがなければセッションを離脱
continue
for message in messages:
print(f"Received: {message}")
await receiver.complete_message(message)
#残りを短い待機時間で処理し、メッセージがなくなった場合には早く離脱できるようにする
async for msg in receiver:
print(f"Received: {message}")
asyncio.run(process_messages())