Azure Python函数并发执行变量被覆盖，邮件内容异常咨询

问题背景 #

在Azure部署的Python Function App中，通过队列触发器触发，从CosmosDB读取指定文档并向用户发送包含文档数据的邮件。当短时间内触发两次函数时，发现第一次执行发送的邮件内容被第二次的新数据覆盖。

函数配置（function.json） #

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "name": "msg",
      "type": "queueTrigger",
      "direction": "in",
      "queueName": "queue_name",
      "connection": "connection_env"
    },
    {
      "type": "cosmosDB",
      "name": "DBfile",
      "databaseName": "db_name",
      "collectionName": "collection_name",
      "connectionStringSetting": "cosmos_db_connection_string",
      "direction": "in",
      "id": "{id}",
      "PartitionKey": "{partitionkey}"
    }
  ]
}

核心代码 #

def main(msg: func.QueueMessage, DBfile: func.DocumentList) -> bool:
    file_raw = DBfile[0].to_json()
    file_json = json.loads(file_raw)
    users = file_json['users']
    for user in users:
        message = {
            "content": {
                "subject": "mail subject",
                "html": json.dumps(file_json['data']),
            },
            "recipients": {
                "to": [{"address": user['emailAddress']}]
            },
            "senderAddress": "<sender@email.com>"
        }

        connection_string_mail = os.getenv("email_connection_string")

        if not connection_string_mail == None:
            email_client = EmailClient.from_connection_string(connection_string_mail)
            email_client.begin_send(message)

假设验证 #

你的假设不成立。Azure Function的每次触发都会创建独立的执行上下文，输入绑定的CosmosDB数据会在函数启动阶段完成读取，存储在当前实例的独立内存空间中。第二次触发读取的新数据，完全不会修改或覆盖第一次执行实例中已加载的DBfile内容。

你遇到的邮件内容异常，更可能是以下原因：

• 异步发送的潜在问题：begin_send()是异步方法，若函数提前执行完毕，局部变量资源可能被回收，导致邮件内容被意外篡改。
• 队列消息重复：两次触发的队列消息携带相同的文档标识，但文档在两次读取间隙已被更新，导致首次发送延迟时使用了后续读取的数据（但此场景下首次读取的旧数据不会被覆盖，仅会出现两次发送不同数据的情况）。

解决方案 #

1. 确保数据独立性 #

显式深拷贝读取到的文档数据，彻底避免任何变量引用导致的意外修改：

import copy

def main(msg: func.QueueMessage, DBfile: func.DocumentList) -> bool:
    file_raw = DBfile[0].to_json()
    file_json = json.loads(file_raw)
    # 深拷贝数据，隔离后续操作对原始读取内容的影响
    file_json = copy.deepcopy(file_json)
    users = file_json['users']
    # 后续代码保持不变

2. 改用同步邮件发送 #

将异步的begin_send替换为同步发送方法，确保邮件内容在发送完成前稳定存在：

# 替换异步发送为同步调用（根据EmailClient实际API调整）
if connection_string_mail:
    email_client = EmailClient.from_connection_string(connection_string_mail)
    email_client.send(message)

3. 实现串行执行（让第二次触发等待） #

如果需要针对同一文档的触发请求串行执行，可通过以下方式实现：

• 乐观并发控制：读取文档时获取_etag字段，处理完成后更新文档时校验_etag，确保同一时间只有一个实例能处理该文档。
• 限制函数并发：在Function App配置中，设置FUNCTIONS_WORKER_PROCESS_COUNT=1，同时将队列触发器的batchSize=1，确保同一时间仅处理一条队列消息。
• 分布式锁：借助Azure Redis Cache或CosmosDB文档锁，函数执行前获取锁，执行完成后释放锁，避免同一文档被并发处理。

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