azure-functions

Azure Functions

使用 Azure Functions 构建和部署无服务器应用程序。涵盖函数应用创建、触发器和绑定配置、部署策略、Python 和 Node.js 实际代码示例，以及生产环境最佳实践。

适用场景

• 你需要可自动缩减至零的事件驱动型计算。
• 你正在构建 API、Webhook 或后台处理管道。
• 你希望按执行次数计费，无需管理服务器。
• 你需要响应 Azure 服务事件（Blob Storage、Service Bus、Cosmos DB 更改）。
• 你正在实现轻量级微服务或计划任务。

前提条件

# Install Azure CLI
curl -sL https://aka.ms/InstallAzureCLIDeb | sudo bash

# Install Azure Functions Core Tools v4
npm install -g azure-functions-core-tools@4

# Verify installation
func --version

# Login
az login
az account set --subscription "my-subscription-id"

# Create supporting resources
az group create --name functions-rg --location eastus

az storage account create \
  --name myfuncstorageacct \
  --resource-group functions-rg \
  --location eastus \
  --sku Standard_LRS

创建函数应用

消耗计划（按执行次数付费）

# Python function app on Consumption plan
az functionapp create \
  --resource-group functions-rg \
  --consumption-plan-location eastus \
  --runtime python \
  --runtime-version 3.11 \
  --functions-version 4 \
  --name myapp-func \
  --storage-account myfuncstorageacct \
  --os-type Linux

# Node.js function app
az functionapp create \
  --resource-group functions-rg \
  --consumption-plan-location eastus \
  --runtime node \
  --runtime-version 20 \
  --functions-version 4 \
  --name myapp-node-func \
  --storage-account myfuncstorageacct \
  --os-type Linux

Premium 计划（VNet 集成，无冷启动）

# Create Premium plan
az functionapp plan create \
  --resource-group functions-rg \
  --name myapp-premium-plan \
  --location eastus \
  --sku EP1 \
  --is-linux true

# Create function app on Premium plan
az functionapp create \
  --resource-group functions-rg \
  --plan myapp-premium-plan \
  --runtime python \
  --runtime-version 3.11 \
  --functions-version 4 \
  --name myapp-premium-func \
  --storage-account myfuncstorageacct

触发器和绑定示例

HTTP 触发器 -- Python

# function_app.py (v2 programming model)
import azure.functions as func
import json
import logging

app = func.FunctionApp(http_auth_level=func.AuthLevel.FUNCTION)

@app.route(route="users/{userId}", methods=["GET"])
def get_user(req: func.HttpRequest) -> func.HttpResponse:
    user_id = req.route_params.get("userId")
    logging.info(f"Fetching user: {user_id}")

    if not user_id:
        return func.HttpResponse(
            json.dumps({"error": "userId is required"}),
            status_code=400,
            mimetype="application/json"
        )

    user = {"id": user_id, "name": "Jane Doe", "email": "[email protected]"}
    return func.HttpResponse(
        json.dumps(user),
        status_code=200,
        mimetype="application/json"
    )

@app.route(route="users", methods=["POST"])
def create_user(req: func.HttpRequest) -> func.HttpResponse:
    try:
        body = req.get_json()
    except ValueError:
        return func.HttpResponse(
            json.dumps({"error": "Invalid JSON"}),
            status_code=400,
            mimetype="application/json"
        )

    logging.info(f"Creating user: {body.get('name')}")
    return func.HttpResponse(
        json.dumps({"id": "new-id", **body}),
        status_code=201,
        mimetype="application/json"
    )

HTTP 触发器 -- Node.js

// src/functions/httpTrigger.js (v4 programming model)
const { app } = require("@azure/functions");

app.http("getUser", {
  methods: ["GET"],
  authLevel: "function",
  route: "users/{userId}",
  handler: async (request, context) => {
    const userId = request.params.userId;
    context.log(`Fetching user: ${userId}`);

    if (!userId) {
      return { status: 400, jsonBody: { error: "userId is required" } };
    }

    const user = { id: userId, name: "Jane Doe", email: "[email protected]" };
    return { status: 200, jsonBody: user };
  },
});

app.http("createUser", {
  methods: ["POST"],
  authLevel: "function",
  route: "users",
  handler: async (request, context) => {
    const body = await request.json();
    context.log(`Creating user: ${body.name}`);

    return { status: 201, jsonBody: { id: "new-id", ...body } };
  },
});

Blob 触发器 -- Python

@app.blob_trigger(arg_name="blob", path="uploads/{name}",
                   connection="AzureWebJobsStorage")
def process_upload(blob: func.InputStream):
    logging.info(f"Processing blob: {blob.name}, Size: {blob.length} bytes")
    content = blob.read()
    # Process file content here

Timer 触发器 -- Python

@app.timer_trigger(schedule="0 */5 * * * *", arg_name="timer",
                    run_on_startup=False)
def cleanup_job(timer: func.TimerRequest):
    if timer.past_due:
        logging.warning("Timer is past due")
    logging.info("Running scheduled cleanup")
    # Cleanup logic here

Service Bus 触发器 -- Python

@app.service_bus_queue_trigger(arg_name="msg", queue_name="orders",
                                connection="ServiceBusConnection")
@app.cosmos_db_output(arg_name="doc", database_name="mydb",
                       container_name="processed-orders",
                       connection="CosmosDBConnection")
def process_order(msg: func.ServiceBusMessage, doc: func.Out[func.Document]):
    order = json.loads(msg.get_body().decode("utf-8"))
    logging.info(f"Processing order: {order['id']}")

    processed = {
        "id": order["id"],
        "status": "processed",
        "items": order["items"],
        "total": sum(item["price"] for item in order["items"])
    }
    doc.set(func.Document.from_dict(processed))

Cosmos DB 更改馈送触发器 -- Python

@app.cosmos_db_trigger_v3(arg_name="documents", database_name="mydb",
                           container_name="orders",
                           connection="CosmosDBConnection",
                           lease_container_name="leases",
                           create_lease_container_if_not_exists=True)
def on_order_change(documents: func.DocumentList):
    for doc in documents:
        logging.info(f"Document changed: {doc['id']}")

本地开发

# Initialize a new Python function project
func init MyFunctionProject --python
cd MyFunctionProject

# Create a new function from template
func new --name HttpExample --template "HTTP trigger" --authlevel function

# Run locally
func start

# Run locally with specific port
func start --port 7072

# Test locally
curl http://localhost:7071/api/HttpExample?name=World

部署

# Deploy using Core Tools
func azure functionapp publish myapp-func

# Deploy with build step for Python
func azure functionapp publish myapp-func --build remote

# Deploy using ZIP package
zip -r function.zip . -x ".git/*" ".venv/*" "__pycache__/*"
az functionapp deployment source config-zip \
  --resource-group functions-rg \
  --name myapp-func \
  --src function.zip

# Deploy via CI/CD with GitHub Actions
az functionapp deployment github-actions add \
  --resource-group functions-rg \
  --name myapp-func \
  --repo "myorg/myrepo" \
  --branch main \
  --runtime python \
  --login-with-github

部署槽

# Create a staging slot
az functionapp deployment slot create \
  --resource-group functions-rg \
  --name myapp-func \
  --slot staging

# Deploy to staging slot
func azure functionapp publish myapp-func --slot staging

# Test staging slot
curl https://myapp-func-staging.azurewebsites.net/api/health

# Swap staging to production
az functionapp deployment slot swap \
  --resource-group functions-rg \
  --name myapp-func \
  --slot staging \
  --target-slot production

# Roll back by swapping again
az functionapp deployment slot swap \
  --resource-group functions-rg \
  --name myapp-func \
  --slot staging \
  --target-slot production

应用程序设置和安全

# Set application settings
az functionapp config appsettings set \
  --resource-group functions-rg \
  --name myapp-func \
  --settings \
    ServiceBusConnection="Endpoint=sb://..." \
    CosmosDBConnection="AccountEndpoint=https://..." \
    CUSTOM_SETTING="my-value"

# Set settings as slot-specific
az functionapp config appsettings set \
  --resource-group functions-rg \
  --name myapp-func \
  --slot-settings \
    ENVIRONMENT="staging"

# Enable managed identity
az functionapp identity assign \
  --resource-group functions-rg \
  --name myapp-func

# Configure CORS
az functionapp cors add \
  --resource-group functions-rg \
  --name myapp-func \
  --allowed-origins "https://myapp.example.com"

# Set minimum TLS version
az functionapp config set \
  --resource-group functions-rg \
  --name myapp-func \
  --min-tls-version 1.2

# Enable Application Insights
az functionapp config appsettings set \
  --resource-group functions-rg \
  --name myapp-func \
  --settings APPINSIGHTS_INSTRUMENTATIONKEY="your-key"

Terraform 配置

resource "azurerm_service_plan" "functions" {
  name                = "myapp-func-plan"
  location            = azurerm_resource_group.main.location
  resource_group_name = azurerm_resource_group.main.name
  os_type             = "Linux"
  sku_name            = "Y1"  # Consumption plan
}

resource "azurerm_linux_function_app" "main" {
  name                       = "myapp-func"
  location                   = azurerm_resource_group.main.location
  resource_group_name        = azurerm_resource_group.main.name
  service_plan_id            = azurerm_service_plan.functions.id
  storage_account_name       = azurerm_storage_account.func.name
  storage_account_access_key = azurerm_storage_account.func.primary_access_key

  identity {
    type = "SystemAssigned"
  }

  site_config {
    application_stack {
      python_version = "3.11"
    }
    cors {
      allowed_origins = ["https://myapp.example.com"]
    }
  }

  app_settings = {
    FUNCTIONS_WORKER_RUNTIME       = "python"
    WEBSITE_RUN_FROM_PACKAGE       = "1"
    APPINSIGHTS_INSTRUMENTATIONKEY = azurerm_application_insights.main.instrumentation_key
  }

  tags = var.tags
}

故障排除

症状	原因	解决方法
冷启动延迟 > 10s	消耗计划的冷启动	使用 Premium 计划（EP1+）或启用 WEBSITE_RUN_FROM_PACKAGE=1
函数未触发	连接字符串配置错误	检查 az functionapp config appsettings list，确认绑定值正确
Python 中出现 ModuleNotFoundError	部署时未安装依赖项	使用 --build remote 标志，或在包中包含 requirements.txt
HTTP 401 未授权	身份验证级别不匹配或缺少函数密钥	确认代码中的身份验证级别符合预期；传递 x-functions-key 标头
Blob 触发器未触发	存储帐户连接错误	确认 AzureWebJobsStorage 指向正确的帐户
Timer 触发器运行两次	Premium 计划上的多个实例	设置 WEBSITE_MAX_DYNAMIC_APPLICATION_SCALE_OUT=1 或使用单例锁
部署槽交换失败	未配置槽设置	确保使用 --slot-settings 标记槽专属设置
内存不足错误	负载过大或内存泄漏	流式处理数据，而不是一次性全部加载；提升计划层级

相关技能

• azure-networking -- 为访问私有资源的 Premium 计划函数配置 VNet 集成。
• azure-sql -- 函数绑定中的数据库连接。
• terraform-azure -- 用于预配函数应用的基础设施即代码。
• arm-templates -- 基于 Bicep 部署函数应用。