如何实现Azure Blob Storage高并发写入？以Azure云函数为例

AI 小北

2026-5-21

Azure云函数场景下Append Blob高并发写入的最佳实践

嘿，针对你在Azure云函数里用Append Blob做高并发写入遇到的数据丢失问题，我来给你梳理下核心原因和可行的解决方案——毕竟Append Blob本身就是为高并发追加场景设计的，只是需要用对姿势。

为什么你第一种方式会丢数据？

当你直接调用appendBlobFromText且不设置访问条件时，看似简单，但高并发场景下会大量出现412前置条件失败的错误（你可以去云函数日志里确认下）。Azure Storage允许Append Blob的并发写入，但如果多个请求同时尝试追加，服务端会返回冲突，而如果你的代码没处理这些错误、没有重试逻辑，这些失败的写入就直接丢了——这就是你40000次写入丢6000次的核心原因。

最佳实现方式

结合你的场景（顺序不重要、云函数高并发），推荐以下几步来解决问题：

1. 给写入操作加上重试逻辑（核心）

不管你用不用访问条件，重试都是必须的。针对412冲突、5xx服务器错误这类可重试的失败，用指数退避策略重新发起写入，这能覆盖绝大多数并发冲突场景。

举个Python新版SDK（azure-storage-blob v12+）的示例：

from azure.storage.blob import AppendBlobClient
from azure.core.exceptions import ResourceModifiedError, ServiceRequestError
import time

def safe_append_to_blob(blob_client, content, max_retries=5):
    retry_count = 0
    while retry_count < max_retries:
        try:
            # 将文本转成字节流，调用append_block
            blob_client.append_block(content.encode('utf-8'))
            return
        except (ResourceModifiedError, ServiceRequestError) as e:
            # ResourceModifiedError对应412冲突，ServiceRequestError对应服务端临时错误
            retry_count += 1
            # 指数退避：第1次等2秒，第2次4秒，以此类推
            time.sleep(2 ** retry_count)
        except Exception as e:
            # 其他不可重试的错误，比如权限问题，直接抛出
            raise
    # 重试次数耗尽仍失败，可根据业务逻辑处理（比如记录日志、入死信队列）
    raise Exception(f"Failed to append after {max_retries} retries")

如果你还在使用旧版SDK（azure-storage-blob v2），逻辑类似，只是捕获的异常类型和API略有不同：

from azure.storage.blob import AppendBlobService
import time

def safe_append_old_sdk(blob_service, container_name, blob_name, content, max_retries=5):
    retry_count = 0
    while retry_count < max_retries:
        try:
            blob_service.append_blob_from_text(container_name, blob_name, content)
            return
        except Exception as e:
            if hasattr(e, 'status_code') and (e.status_code == 412 or 500 <= e.status_code < 600):
                retry_count += 1
                time.sleep(2 ** retry_count)
            else:
                raise
    raise Exception(f"Failed to append after {max_retries} retries")

2. 批量写入减少请求频率

云函数高并发下，每个请求都单独写一次Blob会大幅增加冲突概率。你可以在云函数内部做本地批量缓存（比如攒100条数据再写入），或者用Azure Queue做中间缓冲——让云函数先把数据丢进Queue，再用另一个低并发的函数（或者定时触发器）批量读取Queue数据并写入Append Blob。这种方式能显著降低写入冲突，还能节省存储请求费用。

3. 利用SDK原生重试策略

除了自己写重试逻辑，新版Azure Storage SDK支持配置全局重试策略，比如指数退避。初始化客户端时直接配置，能减少重复代码：

from azure.storage.blob import BlobClientOptions, AppendBlobClient
from azure.core.pipeline.policies import ExponentialBackoffRetry

# 配置指数退避重试：最多重试5次，初始间隔1秒
options = BlobClientOptions(
    retry_policy=ExponentialBackoffRetry(max_retries=5, initial_backoff=1)
)
append_blob_client = AppendBlobClient.from_connection_string(
    "<your-connection-string>",
    "<container-name>",
    "<blob-name>",
    options=options
)

4. 别用访问条件做“强一致性”校验

你提到的第二种方式（带访问条件）其实没必要——指定IfAppendPositionEqual条件看似能保证顺序，但高并发下你的本地获取的append位置会瞬间过时，依然会触发412错误，最终还是要重试。而且每次获取位置还会多一次存储请求，增加开销，完全没必要（毕竟你的场景顺序不重要）。

额外注意事项

控制云函数并发数：如果云函数的最大并发实例太多，会加剧Blob写入冲突。可以在函数应用的配置里适当调低并发数（比如从默认的100调到50），结合批量写入来平衡性能和成功率。
监控写入成功率：用Azure Application Insights或者云函数日志记录每次写入的结果，重点关注失败请求的错误码，方便后续优化。
升级到新版SDK：旧版SDK的重试机制不如新版完善，建议尽快迁移到azure-storage-blob v12+，API更简洁，性能也更好。

内容的提问来源于stack exchange，提问作者sebbulon