# Azure Service Bus

> Use when implementing reliable message processing with Azure Service Bus — choosing between queues and topics, configuring peek-lock settlement, handling dead-lettered messages, or enforcing ordered processing with sessions.

- Skill: `marucie/azure-service-bus` (Agent Skill)
- Install (CLI): `npx skillmds@latest add marucie/azure-service-bus`
- Raw SKILL.md: https://api.skillmd.com/api/skills/marucie/azure-service-bus/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: DevOps & Infra
- Author: MARUCIE (https://skillmd.com/u/marucie)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/marucie/azure-service-bus

---


## 是什么

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

## 怎么用

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

## 架构图

```mermaid
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

```python
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:

| Role | Permission |
|---|---|
| `Azure Service Bus Data Sender` | Send messages only |
| `Azure Service Bus Data Receiver` | Receive and settle messages only |
| `Azure Service Bus Data Owner` | Full control — use only for admin tooling |

## Queues vs Topics/Subscriptions

| Factor | Queue | Topic + Subscriptions |
|---|---|---|
| Consumers | Single consumer group | Multiple independent consumers |
| Fan-out | No | Yes — each subscription gets a copy |
| Filtering | No | Yes — SQL or correlation filter per subscription |
| Ordering | Sessions only | Sessions only |
| Use case | Task queue, work distribution | Event fan-out, pub/sub |

```
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

```python
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

```python
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

```python
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

```python
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.

```python
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

| Method | Effect | When to use |
|---|---|---|
| `complete_message` | Removes from queue | Processing succeeded |
| `abandon_message` | Returns to queue; increments `delivery_count` | Transient failure, will retry |
| `dead_letter_message` | Moves to DLQ with reason | Poison message, max retries exceeded |
| `defer_message` | Parks with sequence number for later retrieval | Out-of-order messages needing dependencies |

### Continuous Consumer (Long-Running)

```python
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

```python
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`.

```python
# 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

```python
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")
```

| Filter type | Performance | Flexibility | Use when |
|---|---|---|---|
| `TrueRuleFilter` | Fast | None | Subscription receives everything (default) |
| `CorrelationRuleFilter` | Fastest | Subject, correlation ID, app properties | Property-based routing |
| `SqlRuleFilter` | Slower | Full SQL expression | Complex conditions across multiple properties |
| `FalseRuleFilter` | Fast | None | Disable subscription without deleting it |

## Retry Configuration

```python
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

```python
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

| Lever | Impact | How |
|---|---|---|
| Tier (Basic vs Standard vs Premium) | High | Basic: queues only, no topics/sessions; Standard: topics + sessions; Premium: dedicated capacity, no throttling |
| Message size | Medium | Standard max 256 KB; Premium max 100 MB — compress large payloads before sending |
| Message TTL | Medium | Set queue/message TTL to avoid accumulating unprocessed messages that waste storage |
| Auto-delete on idle | Low | Set `auto_delete_on_idle` on dev/staging queues to clean up abandoned resources |
| Duplicate detection window | Low | Enable on idempotent queues; deduplicated messages don't count toward throughput billing |

> 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