Azure Service Bus

是什么

Azure Service Bus 是企业级消息中间件，提供队列、主题订阅、死信处理等一整套异步通信能力。 用它的效果是：上下游系统不再强耦合，高峰期流量可被吸收，失败消息有兜底处理路径。

怎么用

1. 先按业务边界规划队列与主题，让生产者与消费者各自只看到自己关心的消息。
2. 在客户端用会话（Session）与分区键控制消息顺序，让有顺序要求的业务不会乱序。
3. 为每条消息设置合理的重试与死信策略，让暂时失败有机会恢复、彻底失败被隔离。
4. 用监控指标观察队列深度与处理延迟，让性能问题在 SLA 失守前被发现。
5. 定期做消息回放演练，让灾难恢复路径处于可用而不是只在文档里。

架构图

flowchart LR
  生产者 --> 消息队列
  消息队列 --> 消费者
  消费者 --> 业务处理
  业务处理 --> 失败兜底
  失败兜底 --> 死信队列
  死信队列 --> 人工介入

Azure Service Bus

Production patterns for Azure Service Bus using the azure-servicebus Python SDK.

When to Activate

• Writing code that imports azure-servicebus or @azure/service-bus
• Choosing between Service Bus queues, topics, and subscriptions
• Implementing reliable message processing with peek-lock and settlement
• Handling dead-lettered messages or poison message scenarios
• Ensuring ordered processing with Service Bus sessions
• Filtering messages per subscriber using SQL or correlation filters
• Scheduling messages for future delivery or implementing deferred processing
• Comparing Azure Service Bus against Azure Storage Queues or Event Hubs

Authentication

from azure.identity import DefaultAzureCredential
from azure.servicebus import ServiceBusClient

credential = DefaultAzureCredential()
client = ServiceBusClient(
    fully_qualified_namespace="myns.servicebus.windows.net",
    credential=credential,
)

# BAD: connection string embeds shared access key
client = ServiceBusClient.from_connection_string(
    "Endpoint=sb://myns.servicebus.windows.net/;SharedAccessKeyName=..."
)

# GOOD: keyless via RBAC role (Azure Service Bus Data Sender / Receiver)
client = ServiceBusClient(
    fully_qualified_namespace=os.environ["SERVICEBUS_NAMESPACE"],
    credential=DefaultAzureCredential(),
)

Always use DefaultAzureCredential and assign the minimum RBAC role:

Queues vs Topics/Subscriptions

Queue:      Producer → [Queue] → Consumer A
Topic:      Producer → [Topic] → [Sub: orders-billing]    → Billing Service
                               → [Sub: orders-shipping]   → Shipping Service
                               → [Sub: orders-analytics]  → Analytics Service

Sending Messages

Single Message

from azure.servicebus import ServiceBusMessage
import json

def send_message(namespace: str, queue_name: str, body: dict, **props) -> None:
    credential = DefaultAzureCredential()
    with ServiceBusClient(namespace, credential) as client:
        with client.get_queue_sender(queue_name) as sender:
            message = ServiceBusMessage(
                json.dumps(body),
                subject=props.get("subject"),
                correlation_id=props.get("correlation_id"),
                message_id=props.get("message_id"),        # deduplication key
                time_to_live=props.get("ttl"),             # timedelta
                application_properties=props.get("properties", {}),
            )
            sender.send_messages(message)

Batch Send

def send_batch(namespace: str, queue_name: str, messages: list[dict]) -> None:
    with ServiceBusClient(namespace, DefaultAzureCredential()) as client:
        with client.get_queue_sender(queue_name) as sender:
            batch = sender.create_message_batch()
            for body in messages:
                try:
                    batch.add_message(ServiceBusMessage(json.dumps(body)))
                except ValueError:
                    # Batch full — send current batch and start a new one
                    sender.send_messages(batch)
                    batch = sender.create_message_batch()
                    batch.add_message(ServiceBusMessage(json.dumps(body)))
            if len(batch):
                sender.send_messages(batch)

Scheduled Messages

from datetime import datetime, timedelta, timezone

def schedule_message(namespace: str, queue_name: str, body: dict, delay: timedelta) -> int:
    enqueue_at = datetime.now(timezone.utc) + delay
    with ServiceBusClient(namespace, DefaultAzureCredential()) as client:
        with client.get_queue_sender(queue_name) as sender:
            seq_numbers = sender.schedule_messages(
                ServiceBusMessage(json.dumps(body)),
                enqueue_at,
            )
            return seq_numbers[0]  # use to cancel with cancel_scheduled_messages()

Sending to a Topic

def publish_event(namespace: str, topic_name: str, event_type: str, payload: dict) -> None:
    with ServiceBusClient(namespace, DefaultAzureCredential()) as client:
        with client.get_topic_sender(topic_name) as sender:
            sender.send_messages(ServiceBusMessage(
                json.dumps(payload),
                subject=event_type,
                application_properties={"event_type": event_type},
            ))

Receiving Messages

Peek-Lock (Recommended)

Locks the message for processing; must be explicitly settled. Ensures at-least-once delivery.

from azure.servicebus import ServiceBusReceiveMode

def process_queue(namespace: str, queue_name: str, handler, max_messages: int = 10) -> None:
    with ServiceBusClient(namespace, DefaultAzureCredential()) as client:
        with client.get_queue_receiver(
            queue_name,
            receive_mode=ServiceBusReceiveMode.PEEK_LOCK,
            max_wait_time=5,
        ) as receiver:
            for msg in receiver.receive_messages(max_message_count=max_messages):
                try:
                    body = json.loads(str(msg))
                    handler(body)
                    receiver.complete_message(msg)      # ack — removes from queue
                except Exception as e:
                    if msg.delivery_count >= 3:
                        receiver.dead_letter_message(   # send to DLQ
                            msg,
                            reason="MaxRetriesExceeded",
                            error_description=str(e),
                        )
                    else:
                        receiver.abandon_message(msg)   # nack — requeues with backoff

Settlement Methods

Continuous Consumer (Long-Running)

import threading

def start_consumer(namespace: str, queue_name: str, handler, stop_event: threading.Event) -> None:
    with ServiceBusClient(namespace, DefaultAzureCredential()) as client:
        with client.get_queue_receiver(queue_name, max_wait_time=5) as receiver:
            while not stop_event.is_set():
                messages = receiver.receive_messages(max_message_count=10, max_wait_time=5)
                if not messages:
                    continue
                for msg in messages:
                    try:
                        handler(json.loads(str(msg)))
                        receiver.complete_message(msg)
                    except Exception as e:
                        logger.exception("message_processing_failed", extra={"delivery_count": msg.delivery_count})
                        receiver.abandon_message(msg)

Dead-Letter Queue

def process_dlq(namespace: str, queue_name: str) -> list[dict]:
    dlq_path = f"{queue_name}/$deadletterqueue"
    dead_letters = []

    with ServiceBusClient(namespace, DefaultAzureCredential()) as client:
        with client.get_queue_receiver(
            dlq_path,
            receive_mode=ServiceBusReceiveMode.PEEK_LOCK,
        ) as receiver:
            for msg in receiver.receive_messages(max_message_count=50):
                dead_letters.append({
                    "body": json.loads(str(msg)),
                    "reason": msg.dead_letter_reason,
                    "description": msg.dead_letter_error_description,
                    "delivery_count": msg.delivery_count,
                    "enqueued_at": msg.enqueued_time_utc,
                })
                receiver.complete_message(msg)  # drain DLQ after inspection

    return dead_letters

Alert when DLQ message count (ActiveMessageCount on the DLQ entity) exceeds zero.

Sessions (Ordered Processing)

Sessions guarantee FIFO ordering for messages with the same session_id. The queue/topic subscription must have requires_session=True.

# Send with session ID (all messages for the same order stay ordered)
def send_with_session(namespace: str, queue_name: str, session_id: str, body: dict) -> None:
    with ServiceBusClient(namespace, DefaultAzureCredential()) as client:
        with client.get_queue_sender(queue_name) as sender:
            sender.send_messages(ServiceBusMessage(
                json.dumps(body),
                session_id=session_id,
            ))

# Receive a specific session
def process_session(namespace: str, queue_name: str, session_id: str, handler) -> None:
    with ServiceBusClient(namespace, DefaultAzureCredential()) as client:
        with client.get_queue_receiver(
            queue_name,
            session_id=session_id,
        ) as receiver:
            for msg in receiver.receive_messages(max_message_count=100):
                handler(json.loads(str(msg)))
                receiver.complete_message(msg)

# Accept the next available session (let Service Bus assign)
def process_next_session(namespace: str, queue_name: str, handler) -> None:
    with ServiceBusClient(namespace, DefaultAzureCredential()) as client:
        with client.get_queue_receiver(
            queue_name,
            session_id=NEXT_AVAILABLE_SESSION,  # from azure.servicebus
        ) as receiver:
            for msg in receiver.receive_messages(max_message_count=100, max_wait_time=10):
                handler(json.loads(str(msg)))
                receiver.complete_message(msg)

Subscription Filters

from azure.servicebus.management import ServiceBusAdministrationClient, SqlRuleFilter, CorrelationRuleFilter

admin = ServiceBusAdministrationClient(
    fully_qualified_namespace="myns.servicebus.windows.net",
    credential=DefaultAzureCredential(),
)

# SQL filter — route by message property
admin.create_rule(
    topic_name="orders",
    subscription_name="orders-eu",
    rule_name="eu-only",
    filter=SqlRuleFilter("Region = 'EU'"),
)

# Correlation filter — cheaper, matches on built-in or application properties
admin.create_rule(
    topic_name="orders",
    subscription_name="orders-priority",
    rule_name="priority-orders",
    filter=CorrelationRuleFilter(
        subject="order.placed",
        application_properties={"priority": "high"},
    ),
)

# Remove the default catch-all rule when adding selective filters
admin.delete_rule("orders", "orders-eu", "$Default")

Retry Configuration

from azure.servicebus import ServiceBusClient
from azure.core.pipeline.policies import RetryPolicy

# SDK-level retry (network/transient errors)
client = ServiceBusClient(
    fully_qualified_namespace="myns.servicebus.windows.net",
    credential=DefaultAzureCredential(),
    retry_total=5,
    retry_backoff_factor=1.5,
    retry_backoff_max=30,
)

# Message-level retry: set max_delivery_count on the queue/subscription
# (default 10 — messages exceeding this are auto-dead-lettered by Service Bus)
admin.update_queue(
    admin.get_queue("my-queue"),
    max_delivery_count=5,
    lock_duration=timedelta(minutes=2),   # must process within this window
)

Error Handling

from azure.servicebus.exceptions import (
    ServiceBusError,
    ServiceBusConnectionError,
    ServiceBusAuthorizationError,
    MessageLockLostError,
    SessionLockLostError,
    MessageAlreadySettled,
)

def safe_receive(receiver, handler, msg) -> None:
    try:
        handler(json.loads(str(msg)))
        receiver.complete_message(msg)
    except MessageLockLostError:
        # Lock expired before we could settle — message will reappear
        logger.warning("lock_expired", message_id=msg.message_id)
    except MessageAlreadySettled:
        # Duplicate settle call — safe to ignore
        pass
    except ServiceBusAuthorizationError:
        # RBAC role missing — fail fast, don't retry
        raise
    except ServiceBusConnectionError as e:
        logger.error("connection_lost", error=str(e))
        raise  # outer loop should reconnect
    except Exception as e:
        logger.exception("handler_failed")
        try:
            receiver.abandon_message(msg)
        except MessageAlreadySettled:
            pass

Cost Controls

See also: event-driven, azure, observability

Red Flags

• Completing a message before the handler finishes — settling with complete_message() before your handler returns means a crash loses the work with no retry opportunity; settle only after successful processing
• No monitoring on the dead-letter queue — DLQ messages represent silently accumulating failures; set an alert on DLQ message count and review DLQ contents after every deployment
• Lock duration shorter than max processing time — if the peek-lock expires before processing completes, the message becomes visible again and gets processed twice; set lock duration to 2–3× your p99 processing time
• Regular receiver used with session-enabled queues — a standard ServiceBusReceiver ignores session grouping and violates ordering guarantees; use accept_next_session() for session-aware delivery
• Catch-all topic subscription with no filters — a TrueRuleFilter subscription on a high-volume topic processes every message; use correlation or SQL filters to subscribe only to relevant message types
• ServiceBusClient recreated per message — each client creation opens a new AMQP connection; create the client once at startup and reuse it across all sends and receives
• Abandoning messages immediately on transient errors — abandoning re-enqueues the message for immediate retry, potentially creating a tight loop; use defer() or back off before abandoning on transient failures

Checklist

• All clients use DefaultAzureCredential — no connection strings or SAS keys in code
• Minimum RBAC role assigned: Data Sender for producers, Data Receiver for consumers
• Receive mode is PEEK_LOCK for all reliable processing — not RECEIVE_AND_DELETE
• All messages explicitly settled: complete, abandon, or dead_letter — no silent drops
• delivery_count checked before settling; poison messages sent to DLQ with reason and description
• DLQ monitored — alert fires when ActiveMessageCount on /$deadletterqueue exceeds 0
• max_delivery_count set on queues/subscriptions (not left at default 10 without review)
• lock_duration set long enough to cover worst-case processing time
• Sessions used for entities requiring ordered or correlated processing
• Subscription $Default rule deleted when selective CorrelationRuleFilter or SqlRuleFilter is added
• SDK client reused across calls — not instantiated per message
• Message TTL set to prevent unbounded queue depth on abandoned consumers