azure function service bus python - 读取一堆消息

Question

有没有机会在我的 python Azure 函数中读取一堆消息（使用 serviceBusTrigger）？

这个问题很简单。我做了很多尝试，但似乎 Azure 函数框架在读取来自 Azure 服务总线的消息后，一次调用触发函数一条消息，无论预取值有多大或设置了任何参数。

我想处理整个消息获取，因为当您必须处理数百万条消息时，一次一条或一次多条会产生影响。

这是我的配置：

function.json

{
    "scriptFile": "main.py",
    "entryPoint": "function_handler",
    "bindings": [{
        "name": "msg",
        "type": "serviceBusTrigger",
        "direction": "in",
        "topicName": "topic-name",
        "subscriptionName": "topic-subscription-name"
    }]
}

ma​​in.py

import azure.functions as func
import json

def function_handler(msg: func.ServiceBusMessage):    
    print(str(msg))

host.json

{
  "version": "2.0",
  "extensionBundle": {
    "id": "Microsoft.Azure.Functions.ExtensionBundle",
    "version": "[1.*, 2.0.0)"
  },
  "extensions": {
    "serviceBus": {
      "prefetchCount": 100,
      "messageHandlerOptions": {
          "autoComplete": true, 
          "maxConcurrentCalls": 6,
          "maxAutoRenewDuration": "00:05:00"
      }
    }
  }
}

Answer 1

我认为您可能需要异步服务总线触发器：

__init__.py

import logging
import asyncio

import azure.functions as func

async def hello():
    await asyncio.sleep(2)

async def main(msg: func.ServiceBusMessage):
    logging.info('Python ServiceBus queue trigger processed message: %s',
                 msg.get_body().decode('utf-8'))
    await asyncio.gather(hello(), hello())
    logging.info(msg.get_body().decode('utf-8')+" is been done.")

host.json

{
  "version": "2.0",
  "logging": {
    "applicationInsights": {
      "samplingSettings": {
        "isEnabled": true,
        "excludedTypes": "Request"
      }
    }
  },
  "extensionBundle": {
    "id": "Microsoft.Azure.Functions.ExtensionBundle",
    "version": "[1.*, 2.0.0)"
  },
  "extensions": {"serviceBus": {"messageHandlerOptions": {"maxConcurrentCalls": 5}}}
}

我可以得到：