从零构建 Serverless 事件驱动工作流：FunctionGraph + 事件总线 EG + 函数工作流 Workflow

当业务从"定时轮询"走向"事件驱动"，从"单函数执行"走向"多步骤编排"，Serverless 的拼图才算真正完整。本文手把手带你用华为云事件总线 EG、FunctionGraph 和函数工作流 Workflow，构建一条可观测、可追溯、可扩展的异步事件处理管道。

一、为什么需要事件驱动工作流

在之前的文章中，我们用 FunctionGraph 实现了定时巡检、API 网关接口和图片处理。但实际业务中，一个事件往往需要多个服务按序协作：

• 用户上传订单 → 校验 → 风控 → 入库 → 通知
• 设备上报告警 → 过滤 → 升级 → 派单 → 归档
• 文件到达 OBS → 解压 → 解析 → 清洗 → 写入数据库

这类场景有三个共同特征：

1. 事件驱动：不是定时触发，而是由某个事件主动推送
2. 多步骤编排：一个函数不够，需要多个步骤按条件执行
3. 异步非阻塞：调用方不关心即时结果，只要求可靠送达

如果只用 FunctionGraph 的单一函数，代码会迅速膨胀成"面条"；如果用消息队列手动编排，运维成本又太高。事件总线 EG + 函数工作流 Workflow 正是为这类场景而生的 Serverless 原生方案。

二、架构全景

我们构建的管道架构如下：

┌─────────────┐    事件投递    ┌──────────────┐   触发执行   ┌─────────────────┐
│  事件源      │ ──────────▶ │  事件总线 EG   │ ─────────▶ │  FunctionGraph   │
│ (自定义应用/  │             │  (事件路由)    │            │  (事件消费函数)   │
│  云服务事件)  │             └──────────────┘            └────────┬────────┘
└─────────────┘                                                   │
                                                                  │ 启动工作流
                                                                  ▼
                                                         ┌─────────────────┐
                                                         │  函数工作流      │
                                                         │  Workflow       │
                                                         │  ┌───────────┐  │
                                                         │  │ Step 1:   │  │
                                                         │  │ 数据校验   │  │
                                                         │  └─────┬─────┘  │
                                                         │        ▼       │
                                                         │  ┌───────────┐  │
                                                         │  │ Step 2:   │  │
                                                         │  │ 业务处理   │  │
                                                         │  └─────┬─────┘  │
                                                         │        ▼       │
                                                         │  ┌───────────┐  │
                                                         │  │ Step 3:   │  │
                                                         │  │ 结果通知   │  │
                                                         │  └───────────┘  │
                                                         └─────────────────┘

各组件职责：

与已有文章的区分：

三、前置准备

3.1 账号与权限

确保你的华为云账号具备以下权限（如不满足，请联系管理员授权）：

• EG Administrator（事件总线管理权限）
• FunctionGraph Administrator（函数工作流管理权限）
• Workflow Administrator（函数工作流管理权限）
• Tenant Guest（OBS 只读权限，用于日志存储）

3.2 环境信息

本文所有操作在 华北-北京四 区域进行。如使用其他区域，请注意事件总线和工作流均需在同一区域创建。

3.3 本地开发环境

# 确认已安装以下工具
python3 --version   # 要求 3.8 及以上
pip --version
# 确认华为云命令行工具（可选，用于快速验证）
hcloud version

四、第一步：创建事件总线

事件总线是整个管道的"交通枢纽"。我们先创建一个自定义事件总线，用于接收来自外部应用的事件。

4.1 在控制台创建

1. 登录华为云控制台，进入 应用集成 > 事件网格 EG
2. 点击 创建事件总线
3. 填写如下配置：

4. 点击 立即创建，等待状态变为"运行中"

4.2 创建事件规则

事件规则决定哪些事件被路由到哪个目标。我们创建一条规则，将所有 order.created 类型的事件路由到 FunctionGraph 函数。

1. 在事件总线详情页，点击 事件规则 > 创建规则
2. 规则配置：

3. 事件模式（JSON）精确匹配：

{
  "source": ["order-service"],
  "type": ["order.created"]
}

4. 目标配置：

5. 点击 立即创建

4.3 事件格式说明

事件总线遵循 CloudEvents 1.0 规范。发送到本总线的事件必须包含以下必填字段：

{
  "id": "事件唯一标识，如 UUID",
  "source": "事件来源，如 order-service",
  "specversion": "1.0",
  "type": "事件类型，如 order.created",
  "time": "事件时间，ISO 8601 格式",
  "datacontenttype": "application/json",
  "data": {
    "orderId": "ORD-20250101-00001",
    "userId": "U-10001",
    "amount": 299.00,
    "items": [
      {"sku": "P-001", "name": "云服务器 s6.large.2", "quantity": 1, "price": 299.00}
    ]
  }
}

注意：source 和 type 字段必须与事件规则中配置的值完全匹配，事件才能被正确路由。

五、第二步：创建事件消费函数

函数作为事件总线的直接消费者，负责接收事件、校验数据、并启动工作流。

5.1 创建函数

1. 进入 函数工作流 FunctionGraph
2. 点击 创建函数，选择 Python 3.9 运行时
3. 基础配置：

4. 点击 创建函数

5.2 委托授权说明

函数需要访问工作流服务，因此必须配置委托。委托的作用是让 FunctionGraph 以安全的方式调用其他云服务，避免在代码中硬编码密钥。

创建委托时，选择以下策略：

• Workflow Invoker（工作流调用权限）
• Tenant Guest（日志写入权限）

5.3 函数代码

将以下代码粘贴到函数编辑器中：

import json
import logging
import os
import uuid

import requests

# 日志配置
logger = logging.getLogger()
logger.setLevel(logging.INFO)

# 工作流端点（从环境变量读取）
WORKFLOW_ENDPOINT = os.environ.get("WORKFLOW_ENDPOINT", "")
WORKFLOW_NAME = os.environ.get("WORKFLOW_NAME", "order-processing-workflow")


def validate_event(event: dict) -> tuple[bool, str]:
    """
    校验事件数据的完整性和合法性。

    Args:
        event: CloudEvents 格式的事件字典

    Returns:
        (是否通过, 错误信息)
    """
    # 1. 检查必填字段
    required_fields = ["id", "source", "type", "data"]
    for field in required_fields:
        if field not in event:
            return False, f"缺少必填字段: {field}"

    # 2. 检查事件类型
    if event.get("type") != "order.created":
        return False, f"不支持的事件类型: {event.get('type')}"

    # 3. 检查 data 内容
    data = event.get("data", {})
    if not isinstance(data, dict):
        return False, "data 字段必须为 JSON 对象"

    order_id = data.get("orderId")
    amount = data.get("amount")
    user_id = data.get("userId")

    if not order_id:
        return False, "缺少订单号 orderId"
    if not user_id:
        return False, "缺少用户 ID userId"
    if amount is None:
        return False, "缺少金额 amount"

    # 4. 金额合法性检查
    try:
        amount_value = float(amount)
        if amount_value <= 0:
            return False, f"金额必须大于 0，当前值: {amount}"
        if amount_value > 1000000:
            return False, f"金额超出单笔上限 1000000，当前值: {amount}"
    except (ValueError, TypeError):
        return False, f"金额格式错误: {amount}"

    # 5. 订单号格式检查
    if not order_id.startswith("ORD-"):
        return False, f"订单号格式错误，应以 ORD- 开头: {order_id}"

    return True, ""


def start_workflow(event_data: dict, correlation_id: str) -> dict:
    """
    调用工作流 API 启动流程。

    Args:
        event_data: 事件数据
        correlation_id: 关联 ID，用于全链路追踪

    Returns:
        工作流执行响应
    """
    url = f"{WORKFLOW_ENDPOINT}/v1/project/{get_project_id()}/workflows/{WORKFLOW_NAME}/executions"

    payload = {
        "input": {
            "eventData": event_data,
            "correlationId": correlation_id,
            "receivedAt": event_data.get("receivedAt", "")
        },
        "name": f"exec-{correlation_id}"
    }

    headers = {
        "Content-Type": "application/json",
        "X-Correlation-ID": correlation_id
    }

    # 使用华为云 SDK 获取鉴权 Token
    # 实际部署时，请使用 IAM 委托方式获取 Token
    token = get_auth_token()
    if token:
        headers["X-Auth-Token"] = token

    response = requests.post(url, json=payload, headers=headers, timeout=10)
    response.raise_for_status()

    return response.json()


def get_project_id() -> str:
    """从环境变量获取项目 ID"""
    return os.environ.get("PROJECT_ID", "")


def get_auth_token() -> str:
    """
    通过委托获取认证 Token。
    实际部署时，函数运行时会自动注入临时 AK/SK。
    """
    # 函数运行时已自动配置委托，此处返回空字符串
    # 实际调用由底层 SDK 处理
    return ""


def handler(event: dict, context):
    """
    函数入口。

    Args:
        event: 事件总线传递的事件数据
        context: 函数运行上下文

    Returns:
        处理结果
    """
    logger.info("收到事件: %s", json.dumps(event, ensure_ascii=False))

    # 1. 校验事件
    is_valid, error_msg = validate_event(event)
    if not is_valid:
        logger.error("事件校验失败: %s", error_msg)
        return {
            "code": "EVENT_INVALID",
            "message": error_msg,
            "eventId": event.get("id", "unknown")
        }

    # 2. 生成关联 ID（用于全链路追踪）
    correlation_id = event.get("id", str(uuid.uuid4()))

    # 3. 启动工作流
    try:
        workflow_response = start_workflow(
            event_data=event.get("data", {}),
            correlation_id=correlation_id
        )
        logger.info("工作流启动成功: %s", json.dumps(workflow_response, ensure_ascii=False))

        return {
            "code": "SUCCESS",
            "message": "工作流已启动",
            "correlationId": correlation_id,
            "workflowExecution": workflow_response.get("execution", {})
        }
    except Exception as e:
        logger.error("工作流启动失败: %s", str(e))
        return {
            "code": "WORKFLOW_START_FAILED",
            "message": str(e),
            "correlationId": correlation_id
        }

5.4 环境变量配置

在函数配置页，添加以下环境变量：

说明：工作流服务地址格式为 workflow.<区域>.myhuaweicloud.com，请根据实际区域替换。

5.5 函数超时与内存配置

为什么超时设 10 秒？ 事件总线对函数响应有超时要求。函数只负责校验和触发工作流，不执行实际业务，10 秒绰绰有余。实际业务逻辑在工作流中异步执行。

六、第三步：创建函数工作流

工作流是本文的核心。它将订单处理拆分为三个步骤：数据校验 → 业务处理 → 结果通知，并支持条件分支和错误重试。

6.1 在工作流控制台创建

1. 进入 函数工作流 Workflow
2. 点击 创建工作流，选择 空白模板
3. 工作流配置：

4. 点击 创建

6.2 工作流定义（DSL）

切换到 DSL 编辑 模式，粘贴以下定义：

version: v2
name: order-processing-workflow
description: 订单事件处理工作流
inputSchema:
  type: object
  properties:
    eventData:
      type: object
      description: 事件数据
    correlationId:
      type: string
      description: 关联 ID
    receivedAt:
      type: string
      description: 事件接收时间
  required:
    - eventData
    - correlationId

start: validateOrder

states:
  - name: validateOrder
    type: task
    resource: "fn://order-validate-function"
    input:
      orderData: ${input.eventData}
      correlationId: ${input.correlationId}
    retry:
      errors:
        - ValidationError
      intervalSeconds: 2
      maxAttempts: 1
      backoffRate: 2
    end: false
    catch:
      - errors:
          - ValidationError
        next: validationFailed

  - name: processOrder
    type: task
    resource: "fn://order-process-function"
    input:
      orderData: ${input.eventData}
      correlationId: ${input.correlationId}
    retry:
      errors:
        - ProcessError
        - DatabaseError
      intervalSeconds: 5
      maxAttempts: 3
      backoffRate: 2
    end: false
    catch:
      - errors:
          - ProcessError
          - DatabaseError
        next: processFailed

  - name: notifySuccess
    type: task
    resource: "fn://order-notify-function"
    input:
      orderData: ${input.eventData}
      correlationId: ${input.correlationId}
      status: "SUCCESS"
    end: true

  - name: notifyFailure
    type: task
    resource: "fn://order-notify-function"
    input:
      orderData: ${input.eventData}
      correlationId: ${input.correlationId}
      status: "FAILED"
      errorMessage: ${input.errorMessage}
    end: true

  - name: validationFailed
    type: fail
    error: "ValidationError"
    message: "订单校验失败"

  - name: processFailed
    type: task
    resource: "fn://order-notify-function"
    input:
      orderData: ${input.eventData}
      correlationId: ${input.correlationId}
      status: "FAILED"
      errorMessage: "订单处理失败"
    end: true

6.3 工作流状态详解

6.4 创建三个子函数

工作流依赖三个 FunctionGraph 函数。逐一创建：

6.4.1 校验函数 order-validate-function

import json
import logging

logger = logging.getLogger()
logger.setLevel(logging.INFO)


def handler(event: dict, context):
    """订单校验函数"""
    logger.info("校验函数收到: %s", json.dumps(event, ensure_ascii=False))

    order_data = event.get("orderData", {})
    correlation_id = event.get("correlationId", "")

    # 校验规则
    errors = []

    # 金额校验
    amount = order_data.get("amount", 0)
    if amount <= 0:
        errors.append("金额必须大于 0")
    if amount > 1000000:
        errors.append("金额超出单笔上限")

    # 商品校验
    items = order_data.get("items", [])
    if not items:
        errors.append("订单商品列表不能为空")

    # 用户 ID 校验
    user_id = order_data.get("userId", "")
    if not user_id or not user_id.startswith("U-"):
        errors.append("用户 ID 格式错误")

    if errors:
        error_msg = "; ".join(errors)
        logger.error("校验失败: %s", error_msg)
        raise Exception(json.dumps({
            "error": "ValidationError",
            "message": error_msg,
            "correlationId": correlation_id
        }, ensure_ascii=False))

    return {
        "valid": True,
        "correlationId": correlation_id,
        "orderData": order_data
    }

6.4.2 业务处理函数 order-process-function

import json
import logging
import time

logger = logging.getLogger()
logger.setLevel(logging.INFO)


def handler(event: dict, context):
    """订单业务处理函数"""
    logger.info("处理函数收到: %s", json.dumps(event, ensure_ascii=False))

    order_data = event.get("orderData", {})
    correlation_id = event.get("correlationId", "")

    # 模拟业务处理（实际场景中可调用数据库、第三方服务等）
    time.sleep(0.5)  # 模拟处理耗时

    # 计算订单总额
    items = order_data.get("items", [])
    total = sum(item.get("price", 0) * item.get("quantity", 1) for item in items)

    # 生成处理结果
    result = {
        "orderId": order_data.get("orderId"),
        "userId": order_data.get("userId"),
        "totalAmount": total,
        "itemCount": len(items),
        "processedAt": time.strftime("%Y-%m-%dT%H:%M:%S"),
        "status": "PROCESSED",
        "correlationId": correlation_id
    }

    logger.info("处理完成: %s", json.dumps(result, ensure_ascii=False))

    return result

6.4.3 通知函数 order-notify-function

import json
import logging

logger = logging.getLogger()
logger.setLevel(logging.INFO)


def handler(event: dict, context):
    """通知函数"""
    logger.info("通知函数收到: %s", json.dumps(event, ensure_ascii=False))

    order_data = event.get("orderData", {})
    correlation_id = event.get("correlationId", "")
    status = event.get("status", "")
    error_message = event.get("errorMessage", "")

    # 构建通知内容
    if status == "SUCCESS":
        title = "订单处理成功"
        content = f"""
订单号: {order_data.get('orderId')}
用户: {order_data.get('userId')}
金额: {order_data.get('amount')}
商品数: {len(order_data.get('items', []))}
状态: 处理完成
        """.strip()
    else:
        title = "订单处理失败"
        content = f"""
订单号: {order_data.get('orderId')}
用户: {order_data.get('userId')}
错误: {error_message}
        """.strip()

    # 实际场景中可调用 SMN 发送短信/邮件/站内信
    # 此处仅记录日志
    logger.info("通知 - 标题: %s, 内容: %s", title, content)

    return {
        "notified": True,
        "title": title,
        "correlationId": correlation_id
    }

6.5 工作流权限配置

工作流需要调用 FunctionGraph 函数，因此必须配置委托：

1. 在工作流详情页，点击 权限配置
2. 创建委托 workflow-functiongraph-trust
3. 添加策略：FunctionGraph Invoker

七、第四步：发送测试事件

现在所有组件已就绪，我们发送一条测试事件来验证整个管道。

7.1 通过控制台发送

1. 进入事件总线 order-event-bus 详情页
2. 点击 事件管理 > 发送事件
3. 填写事件内容：

{
  "id": "evt-20250101-00001",
  "source": "order-service",
  "specversion": "1.0",
  "type": "order.created",
  "time": "2025-01-01T10:00:00Z",
  "datacontenttype": "application/json",
  "data": {
    "orderId": "ORD-20250101-00001",
    "userId": "U-10001",
    "amount": 299.00,
    "items": [
      {
        "sku": "P-001",
        "name": "云服务器 s6.large.2",
        "quantity": 1,
        "price": 299.00
      }
    ]
  }
}

4. 点击 发送

7.2 通过 API 发送（自动化测试）

# 替换为你的实际项目 ID 和区域
PROJECT_ID="your-project-id"
REGION="cn-north-4"
EVENT_BUS_NAME="order-event-bus"

# 获取 Token（实际部署时请使用 IAM 委托方式）
# 此处假设已配置好环境变量或使用命令行工具

# 发送事件（请将 <API_ENDPOINT> 替换为实际的事件总线 API 地址）
curl -X POST \
  "<API_ENDPOINT>/v1/${PROJECT_ID}/event-buses/${EVENT_BUS_NAME}/events" \
  -H "Content-Type: application/json" \
  -H "X-Auth-Token: ${TOKEN}" \
  -d '[
    {
      "id": "evt-20250101-00002",
      "source": "order-service",
      "specversion": "1.0",
      "type": "order.created",
      "time": "2025-01-01T10:00:00Z",
      "datacontenttype": "application/json",
      "data": {
        "orderId": "ORD-20250101-00002",
        "userId": "U-10002",
        "amount": 599.00,
        "items": [
          {
            "sku": "P-002",
            "name": "云数据库 MySQL 1C2G",
            "quantity": 1,
            "price": 599.00
          }
        ]
      }
    }
  ]'

说明：API 端点格式为 eg.<区域>.myhuaweicloud.com，请根据实际区域替换。

7.3 验证执行结果

验证事件总线

1. 进入事件总线详情页
2. 点击 事件管理 > 事件投递记录
3. 确认事件状态为"投递成功"

验证函数执行

1. 进入函数 order-event-consumer 详情页
2. 点击 监控 > 日志
3. 确认看到"收到事件"和"工作流启动成功"的日志

验证工作流执行

1. 进入工作流 order-processing-workflow 详情页
2. 点击 执行记录
3. 确认看到一条状态为"成功"的执行记录
4. 点击执行记录，查看各步骤状态：

validateOrder  ✓ 完成
processOrder   ✓ 完成
notifySuccess  ✓ 完成

7.4 异常场景测试

为了验证工作流的错误处理能力，我们发送一条非法事件：

{
  "id": "evt-20250101-00003",
  "source": "order-service",
  "specversion": "1.0",
  "type": "order.created",
  "time": "2025-01-01T10:00:00Z",
  "datacontenttype": "application/json",
  "data": {
    "orderId": "ORD-20250101-00003",
    "userId": "U-10003",
    "amount": -100,
    "items": []
  }
}

预期结果：

• 校验函数返回 ValidationError
• 工作流进入 validationFailed 状态
• 工作流执行记录状态为"失败"
• 日志中记录详细错误信息

八、全链路追踪

在分布式系统中，追踪一个事件从发起到完成的完整路径至关重要。我们通过 关联 ID（Correlation ID） 实现全链路追踪。

8.1 追踪原理

事件生成 → 关联 ID 注入 → 事件总线 → 函数 → 工作流 → 各步骤
   ↑                                           ↓
   └──────────── 日志中携带相同 ID ←─────────────┘

8.2 在日志中追踪

所有服务的日志都包含 correlationId 字段。在 LTS 日志服务中，使用以下查询语句追踪单个事件：

correlationId: "evt-20250101-00001"

这将返回该事件在所有服务中的日志记录，按时间排序即可看到完整链路。

8.3 在控制台追踪

1. 进入工作流执行记录
2. 点击目标执行记录
3. 查看 执行详情 页，包含：
   • 各步骤的输入/输出
   • 各步骤的执行时间
   • 错误堆栈（如有）
   • 关联 ID

九、生产环境最佳实践

9.1 错误重试策略

工作流内置了指数退避重试机制。配置建议：

9.2 幂等性设计

事件可能被重复投递（至少一次投递语义），因此所有函数必须实现幂等性：

def handler(event: dict, context):
    correlation_id = event.get("correlationId")

    # 检查是否已处理
    if is_processed(correlation_id):
        return {"status": "already_processed"}

    # 执行处理
    result = process_event(event)

    # 标记已处理
    mark_processed(correlation_id)

    return result

9.3 安全加固

1. 事件签名验证：在事件源端对事件进行签名，函数端验证签名
2. 最小权限原则：函数委托仅授予必要的服务调用权限
3. 网络隔离：如处理敏感数据，将函数部署在 VPC 内
4. 敏感数据脱敏：日志中不记录完整身份证号、银行卡号等

9.4 成本优化

9.5 监控告警

建议配置以下监控指标：

十、常见问题

Q1：事件总线和工作流必须在同一区域吗？

是的。 事件总线规则中配置的函数目标必须与函数、工作流处于同一区域。跨区域调用会增加延迟且可能产生额外费用。

Q2：工作流执行失败会重试吗？

会。 工作流内置了错误捕获和重试机制。在 DSL 中通过 retry 字段配置重试策略。如果所有重试都失败，会进入 catch 分支或 fail 状态。

Q3：如何保证事件不丢失？

事件总线提供至少一次投递保证。如需精确一次语义，需要在消费端实现幂等性。对于关键业务，建议在事件数据中增加唯一 ID，并在消费端做去重。

Q4：工作流支持哪些状态类型？

Q5：函数执行超时怎么办？

函数超时后，事件总线会重试投递（最多 3 次）。如果函数需要长时间执行，应改为：

1. 函数快速返回"已接收"
2. 将实际处理逻辑放在工作流中
3. 工作流中的 Task 状态可配置更长超时（最长 1 年）

十一、总结

本文从零构建了一条完整的 Serverless 事件驱动工作流管道，涵盖以下核心知识点：

1. 事件总线 EG：作为事件中枢，统一接收、路由、投递事件
2. FunctionGraph：作为事件消费者，负责数据校验和工作流触发
3. 函数工作流 Workflow：作为流程编排引擎，实现多步骤、条件分支、错误重试
4. 全链路追踪：通过关联 ID 实现跨服务日志串联

这套架构相比传统的"定时轮询 + 单体函数"方案，具有显著优势：

• 实时性：事件驱动，毫秒级响应
• 可维护性：工作流可视化，逻辑清晰
• 可扩展性：新增事件类型只需添加规则
• 成本可控：按调用计费，空闲零成本

下一步，你可以尝试：

• 接入更多事件源（云服务事件、第三方 Webhook）
• 添加并行处理步骤
• 集成 SMN 实现多渠道通知
• 使用 LTS 实现日志分析和告警

本文由华为云开发者社区发布，欢迎转载分享，请注明出处。