如何用Python队列触发Azure Function将Azure Queue消息存入Blob存储

两种方案都比你之前用Azure Logic App的成本低很多，具体差异和适配场景如下：

• 方案一：直接修改现有HTTP触发Function同步写入Blob
  • 优势：架构最精简，无额外Function调用成本，写入延迟最低，不需要额外维护另一套Function资源
  • 适用场景：写入Blob的逻辑简单、不需要单独重试、对HTTP请求响应延迟容忍度较高，单分钟请求量小于1000次的场景
  • 劣势：Blob写入失败会直接影响主链路的接口响应，甚至导致队列写入也失败，相当于增加了主业务的故障点；如果Blob写入耗时较长会拉长接口的响应时间
• 方案二：新增独立队列触发的Function写入Blob
  • 优势：和主业务链路完全解耦，Blob写入失败不会影响HTTP接口的正常响应和队列写入，自带队列重试机制，可靠性更高；后续如果要扩展写入Blob的逻辑不需要修改主接口代码，迭代更安全
  • 适用场景：对HTTP接口响应延迟要求高、写入Blob逻辑复杂可能需要重试、后续Blob写入逻辑会频繁迭代的场景

选型建议：如果你的业务对接口响应时间要求不高，也不需要单独的Blob写入重试策略，选方案一更省事；否则选方案二稳定性更好。

方案一（直接修改现有HTTP Function）实现步骤 #

1. 先修改你现有HTTP Function的function.json配置，新增Blob输出绑定，示例如下：

{
  "type": "blob",
  "direction": "out",
  "name": "outputblob",
  "path": "你要存储的Blob容器名/{rand-guid}.json",
  "connection": "AzureWebJobsStorage"
}

配置里的AzureWebJobsStorage默认读取你Function App已配置的存储账户连接串，如果要存到独立的Blob存储账户，替换为你自定义的连接串配置名即可；{rand-guid}是系统内置参数，会自动生成唯一ID作为文件名避免覆盖，你也可以替换为时间戳规则{datetime:yyyyMMdd-HHmmss-fff}。

2. 修改后的Python代码如下：

import logging
import azure.functions as func
import json
def main(req: func.HttpRequest, queueclient: func.Out[func.QueueMessage], outputblob: func.Out[bytes]) -> str:
    logging.info('Python HTTP trigger function processed a request.')
    print(req.method)

    if req.method == 'GET': 
        queueclient.set('string')
        return func.HttpResponse(body='------------------------', status_code=200)

    elif req.method =='POST':
        try:
            req_body = req.get_json()
        except ValueError:
            return func.HttpResponse( 
                body="<h1>请求体格式错误</h1>",
                status_code=400
            )
        else:
            sharedSecret = req_body.get('secret')
        if sharedSecret =='':
            req_body_str = json.dumps(req_body)
            # 原有写入队列逻辑
            queueclient.set(req_body_str)
            # 新增同步写入Blob逻辑
            outputblob.set(req_body_str.encode('utf-8'))
            return func.HttpResponse(body=req_body_str, status_code=201, headers={'content-Type': 'application/json'})
        else:
            return func.HttpResponse( 
                body="<h1>请在请求体中添加sharedsecret字符串</h1>",
                status_code=401
            )

方案二（新增队列触发独立Function）实现步骤 #

1. 新建队列触发Function，配置function.json如下：

{
  "scriptFile": "__init__.py",
  "bindings": [
    {
      "name": "msg",
      "type": "queueTrigger",
      "direction": "in",
      "queueName": "你的现有队列名称",
      "connection": "AzureWebJobsStorage"
    },
    {
      "type": "blob",
      "direction": "out",
      "name": "outputblob",
      "path": "你要存储的Blob容器名/{rand-guid}.json",
      "connection": "AzureWebJobsStorage"
    }
  ]
}

2. 队列触发Function的__init__.py代码如下：

import logging
import azure.functions as func

def main(msg: func.QueueMessage, outputblob: func.Out[bytes]) -> None:
    logging.info('队列触发函数收到消息，内容为: %s', msg.get_body().decode('utf-8'))
    # 直接将队列消息写入Blob存储
    outputblob.set(msg.get_body())

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