Azure Blob存储文件锁实现：Python双进程读写方案问询

作为经常处理Azure Blob并发场景的开发者，我来给你梳理下怎么实现这个锁机制——Azure Blob本身没有像本地文件那样的原生锁，但它提供的**租约（Lease）**机制完美适配你的双进程安全读写需求。租约本质是给Blob加一个临时的"锁定标记"，只有持有租约ID的客户端才能修改/删除Blob，其他客户端只能等待租约释放或过期。

核心思路适配你的场景 #

你的需求可以拆解为以下流程：

• 进程尝试获取目标Blob的租约：
  • 如果Blob不存在：创建Blob后立即获取租约，执行Docker命令、写入结果，最后释放租约
  • 如果Blob存在但无租约：获取租约后执行读取操作（或按需写入）
  • 如果Blob存在且有租约：等待一段时间后重试，直到租约释放/过期

具体实现步骤（Python） #

1. 安装依赖 #

确保你安装了Azure Blob的Python SDK：

pip install azure-storage-blob

2. 完整代码示例 #

下面是适配你需求的脚本，包含租约获取、重试逻辑、读写操作：

import time
from azure.storage.blob import BlobServiceClient, BlobLeaseClient

# 替换成你的Azure Blob连接字符串和容器/Blob名称
CONNECTION_STRING = "your_blob_connection_string"
CONTAINER_NAME = "your_container_name"
BLOB_NAME = "target_file.txt"
# 租约有效期（秒），建议设置为你执行Docker命令的最长预期时间，避免进程崩溃后长期锁定
LEASE_DURATION = 300
# 重试间隔（秒）
RETRY_INTERVAL = 5

def get_blob_lease(blob_client):
    """尝试获取Blob的租约，返回租约客户端或None"""
    try:
        # 创建租约客户端，尝试获取独占租约
        lease_client = BlobLeaseClient(blob_client)
        lease_id = lease_client.acquire(lease_duration=LEASE_DURATION)
        print(f"成功获取租约，ID: {lease_id}")
        return lease_client
    except Exception as e:
        # 租约被其他进程持有会抛出异常，这里简化处理
        print(f"无法获取租约，重试中... 原因: {str(e)}")
        return None

def main():
    # 初始化Blob服务客户端
    blob_service_client = BlobServiceClient.from_connection_string(CONNECTION_STRING)
    container_client = blob_service_client.get_container_client(CONTAINER_NAME)
    blob_client = container_client.get_blob_client(BLOB_NAME)

    lease_client = None
    try:
        # 循环尝试获取租约，直到成功
        while lease_client is None:
            lease_client = get_blob_lease(blob_client)
            if lease_client is None:
                time.sleep(RETRY_INTERVAL)

        # 检查Blob是否存在
        if not blob_client.exists():
            print("目标文件不存在，创建并锁定后执行操作")
            # 创建空Blob（如果需要可以先写入初始内容）
            blob_client.upload_blob(b"", overwrite=True)
            # 执行你的Docker命令（替换成实际命令）
            # docker_command_result = os.popen("your_docker_command").read()
            docker_command_result = "Docker执行结果示例"
            # 写入结果到Blob
            blob_client.upload_blob(docker_command_result, overwrite=True)
            print("操作完成，写入结果到Blob")
        else:
            print("目标文件存在，读取内容")
            # 读取Blob内容（因为我们持有租约，此时不会有其他进程修改）
            blob_content = blob_client.download_blob().readall().decode("utf-8")
            print(f"读取到内容: {blob_content}")

    finally:
        # 无论成功失败，都尝试释放租约（避免不必要的锁定）
        if lease_client is not None:
            try:
                lease_client.release()
                print("已释放租约")
            except Exception as e:
                print(f"释放租约失败: {str(e)}")

if __name__ == "__main__":
    main()

关键细节说明 #

• 租约类型：上面用的是有限期租约（lease_duration设置为300秒），如果进程意外崩溃，租约会自动过期，不会导致Blob永久锁定。如果需要更严格的独占控制，也可以用无限期租约，但必须确保进程正常释放。
• 重试逻辑：用循环+time.sleep实现简单的重试，你可以根据需求调整重试次数或改用指数退避策略（比如每次重试间隔翻倍）。
• 读取的安全性：如果你的场景要求读取时绝对不能有其他进程写入，必须在读取前获取租约；如果只是读取最新内容，不需要强一致性，也可以跳过租约直接读取，但要注意此时可能有写入冲突。
• 异常处理：示例中简化了异常处理，实际生产环境建议捕获Azure SDK的具体异常（比如LeaseAlreadyPresentError、ResourceNotFoundError等），让逻辑更健壮。

额外建议 #

• 可以在Blob的元数据中记录锁定进程的标识，方便调试，但这不是必须的。
• 如果后续扩展到多进程场景，租约机制同样适用，只需要保持重试逻辑即可。

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