如何基于Python v2编程模型本地开发EventHub触发的Azure Functions

1. 能否以合理成本本地开发？ #

完全可以，两种低成本方案任选：

• Azure Event Hubs免费层：零成本起步，提供每日100万条消息、1GB数据传输额度，完全覆盖开发测试需求，超出额度后费用极低。
• 本地容器模拟：通过Docker运行兼容Event Hubs协议的消息服务（如Apache Kafka），实现离线本地开发，全程零成本，仅需额外配置容器环境。

2. 可运行示例及Thunder Client说明 #

修正后的Python触发函数代码 #

基于你提供的示例，替换占位符为实际配置值：

import azure.functions as func
import logging

app = func.FunctionApp()

@app.function_name(name="EventHubTrigger")
@app.event_hub_message_trigger(
    arg_name="myhub", 
    event_hub_name="your-event-hub-name",
    connection="EventHubConnectionString"
) 
def test_function(myhub: func.EventHubEvent):
    logging.info('Python EventHub trigger processed an event: %s',
                myhub.get_body().decode('utf-8'))

测试方式 #

Event Hub触发函数是被动监听消息的，Thunder Client无法直接触发（它仅用于HTTP请求调试）。需通过以下方式向Event Hub发送消息触发函数：

• Azure Portal手动发送：登录Azure Portal进入你的Event Hub实例，找到「发送消息」面板，输入文本消息后发送，本地运行的函数会立即捕获并打印日志。
• Python脚本批量发送：用Azure Event Hubs SDK编写简单发送脚本：

from azure.eventhub import EventHubProducerClient, EventData

connection_str = "你的Event Hub连接字符串"
event_hub_name = "your-event-hub-name"

producer = EventHubProducerClient.from_connection_string(connection_str, event_hub_name=event_hub_name)
with producer:
    event_data_batch = producer.create_batch()
    event_data_batch.add(EventData('测试Event Hub触发消息'))
    producer.send_batch(event_data_batch)
print("消息发送成功")

运行脚本后，本地函数的终端会输出处理日志。

3. host.json和local.settings.json配置 #

local.settings.json（本地运行专属配置，需加入.gitignore） #

{
  "IsEncrypted": false,
  "Values": {
    "AzureWebJobsStorage": "UseDevelopmentStorage=true", // 本地存储模拟器，用于函数运行时状态管理
    "FUNCTIONS_WORKER_RUNTIME": "python",
    "EventHubConnectionString": "你的Event Hub连接字符串" // 需与代码中connection参数名一致
  }
}

注：连接字符串可从Azure Portal的Event Hub命名空间「共享访问策略」中获取，选择具备「发送」「监听」权限的策略，复制完整连接字符串。

host.json（函数应用全局配置，最小可运行示例） #

{
  "version": "2.0",
  "logging": {
    "applicationInsights": {
      "samplingSettings": {
        "isEnabled": true,
        "excludedTypes": "Request"
      }
    }
  },
  "extensions": {
    "eventHubs": {
      "batchCheckpointFrequency": 1,
      "maxBatchSize": 10
    }
  }
}

注：eventHubs节点为可选配置，用于调整批量处理、 checkpoint频率等，默认配置也可正常运行。

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