本文介绍如何为 Azure Event Hub 触发的函数设计健壮的异常处理机制，通过自定义装饰器和异常链（raise ... from e）精准捕获任意异常，同时完整保留原始 JSON 消息体、异常类型及堆栈信息，便于后续重试、诊断或转发至死信队列。
本文介绍如何为 azure event hub 触发的函数设计健壮的异常处理机制，通过自定义装饰器和异常链（`raise ... from e`）精准捕获任意异常，同时完整保留原始 json 消息体、异常类型及堆栈信息，便于后续重试、诊断或转发至死信队列。
在构建高可靠性的事件驱动服务（如 Azure Functions + Event Hub）时，一个核心挑战是：当业务逻辑抛出异常时，既要不丢失原始输入消息（即 eventHubMessage: str），又要完整保留原始异常的类型、消息和调用栈，以便进行精准诊断、结构化告警或生成带上下文的错误事件。
直接使用裸 except Exception as e: 并重新 raise 会丢失原始异常链；而简单记录日志又无法在上层统一拦截并注入原始消息。理想的方案是：将原始消息作为上下文绑定到异常中，并利用 Python 的异常链机制（raise NewException(...) from original_exc）显式保留因果关系。
✅ 推荐方案：带上下文的异常装饰器 + 异常链
以下是一个专为 Event Hub 触发器设计的可复用装饰器，它自动捕获所有异常，并将原始消息、异常对象、时间戳封装进自定义异常：
from functools import wraps
import logging
import json
from typing import Any, Dict, Optional
logger = logging.getLogger(__name__)
class EventProcessingError(Exception):
"""自定义异常，携带原始事件与原始异常"""
def __init__(
self,
original_message: str,
original_exception: Exception,
context: Optional[Dict[str, Any]] = None
):
self.original_message = original_message
self.original_exception = original_exception
self.context = context or {}
# 构造可读的 detail，包含原始异常消息和类型
exc_type = type(original_exception).__name__
exc_msg = str(original_exception)
super().__init__(f"[{exc_type}] {exc_msg} | Raw message length: {len(original_message)} chars")
def handle_event_exceptions(func):
"""
装饰器：捕获函数内所有异常，包装为 EventProcessingError 并保留原始消息与异常链
"""
@wraps(func)
async def wrapper(eventHubMessage: str, *args, **kwargs):
try:
return await func(eventHubMessage, *args, **kwargs)
except Exception as e:
# 关键：使用 'from e' 保留原始异常链
raise EventProcessingError(
original_message=eventHubMessage,
original_exception=e,
context={"function": func.__name__}
) from e
return wrapper
? 在 main 函数中应用装饰器
修改你的入口函数，直接应用该装饰器：
@handle_event_exceptions
async def main(eventHubMessage: str):
try:
data = json.loads(eventHubMessage)
logging.info(f"Processing message: {data}")
if isinstance(data, list):
for record in data:
validate_request(record)
await process_request(record)
else:
validate_request(data)
await process_request(data)
logging.info("All requests processed successfully.")
except json.JSONDecodeError as e:
# 显式转为 ValueError，保持语义清晰（也可直接让装饰器捕获）
raise ValueError(f"Invalid JSON: {e}") from e
except ValueError as ve:
raise ve # 让装饰器统一包装
except Exception as e:
raise e # 所有异常最终由装饰器捕获并增强
? 后续处理：发送结构化错误事件（关键实践）
在顶层异常处理器（例如 Azure Function 的主入口或全局异常中间件）中，你可以安全地提取原始消息与错误详情：
# 在 main() 调用处增加顶层 try-catch（Azure Function 推荐方式）
async def main_wrapper(eventHubMessage: str):
try:
await main(eventHubMessage)
except EventProcessingError as e:
# ✅ 此刻你拥有：
# - e.original_message → 完整原始 JSON 字符串
# - e.original_exception → 原始异常对象（可 .__cause__ 或 .args 访问）
# - e.context → 额外上下文（如函数名）
# 示例：构造错误事件并发送到 Dead Letter Topic
error_event = {
"original_message": e.original_message,
"error_type": type(e.original_exception).__name__,
"error_message": str(e.original_exception),
"timestamp": datetime.utcnow().isoformat(),
"function": e.context.get("function", "unknown"),
"traceback": "".join(
traceback.format_exception(type(e.original_exception), e.original_exception, e.original_exception.__traceback__)
)
}
# TODO: 发送 error_event 到另一个 Event Hub topic 或 Storage Queue
logging.error(f"Failed to process event. Sending to DLQ: {error_event['error_type']} - {error_event['error_message']}")
await send_to_dead_letter_topic(json.dumps(error_event))
⚠️ 注意事项与最佳实践
-
不要吞掉异常：避免
except Exception: pass或仅logging.error()而不raise—— 这会导致事件被静默丢弃，违反至少一次投递语义。 -
JSON 解析失败需明确处理：
json.JSONDecodeError是常见首错点，建议在装饰器前单独捕获并转为ValueError，确保语义一致。 -
日志级别要合理：对
EventProcessingError使用logging.exception()可自动打印完整 traceback；对业务校验错误（如字段缺失）用logging.warning()即可。 -
Azure Function 重试策略配合：在
function.json中配置"maxRetryCount": 3，结合死信策略，实现弹性容错。 - 性能考虑：装饰器本身无显著开销；但序列化完整 traceback 到 JSON 时注意长度（可截断或异步写入）。
通过该方案，你不再需要在每个 try/except 块中手动传递 eventHubMessage，所有异常都自动携带上下文，既满足可观测性要求，也为构建可审计、可重放、可追踪的事件流水线打下坚实基础。
