Temporal 工作流引擎深度实战：从零构建弹性、可恢复的分布式事务编排系统

引言：为什么分布式系统需要工作流引擎？

在微服务架构中，一个业务操作往往需要跨越多个服务：创建订单需要调用库存服务扣减库存、调用支付服务扣款、调用通知服务发送确认。当任意一步失败时，如何保证数据一致性？传统的分布式事务（2PC/TCC）复杂且脆弱，而基于消息队列的最终一致性方案又难以追踪和调试整条链路。

Temporal（前身为 Uber Cadence）是一款开源的持久化工作流编排引擎，它将分布式事务的状态编排转化为普通代码中的控制流，同时自动处理失败重试、超时补偿、状态持久化和可视化管理。本文将深入剖析 Temporal 的核心架构，并通过完整的生产级实战案例，展示如何用 Temporal 替代传统 Saga 编排、消息队列和定时任务。

一、Temporal 核心架构解析

1.1 四层组件拓扑

Temporal 采用客户端-服务端分离架构，主要由四个核心层组成：

• Client SDK：嵌入应用代码的客户端库，提供 Workflow/Activity 定义 API（支持 Java/Go/TS/Python/PHP/.NET 等多语言）
• Temporal Service：核心服务端，包含 Frontend Gateway、History Service、Matching Service、Worker Service 四个内部服务
• Persistence Layer：底层持久化存储，支持 Cassandra、PostgreSQL、MySQL 等，存储所有工作流状态事件
• Web UI：可视化控制台，提供工作流执行历史查看、状态搜索、调试和时间线可视化

1.2 执行模型：Workflow + Activity

Temporal 的执行模型基于两个核心抽象：

• Workflow（工作流）：定义业务流程编排逻辑，必须具有确定性（deterministic），Temporal 通过事件溯源（Event Sourcing）保证执行可重现
• Activity（活动）：执行具体业务逻辑的单元（如调用外部API、操作数据库、发送消息），支持自动重试和超时控制

1.3 事件溯源与确定性保证

Temporal 不直接存储工作流状态，而是记录所有事件（Event）。每当工作流执行到需要等待的地方（如 Activity 完成、定时器触发、信号到达），Temporal 暂停工作流执行并保存事件。恢复时，SDK 重放所有事件让工作流恢复到暂停点，然后继续执行。这种机制保证了：

• 进程崩溃后工作流自动恢复
• 任意时间点的执行状态可审计
• 开发者用同步代码编写异步流程

二、SDK 编程模型深度实战（TypeScript）

2.1 项目初始化与连接配置

// src/worker.ts
import { Worker, NativeConnection } from '@temporalio/worker';
import * as activities from './activities';

async function run() {
  const connection = await NativeConnection.connect({
    address: 'localhost:7233',
  });

  const worker = await Worker.create({
    connection,
    namespace: 'default',
    taskQueue: 'order-saga-queue',
    workflowsPath: require.resolve('./workflows'),
    activities,
    maxConcurrentActivityTaskExecutions: 100,
    maxConcurrentWorkflowTaskExecutions: 50,
  });

  await worker.run();
  console.log('Temporal Worker started successfully');
}

run().catch((err) => {
  console.error(err);
  process.exit(1);
});

2.2 定义 Activity：业务逻辑单元

// src/activities.ts
import { Context } from '@temporalio/activity';
import axios from 'axios';

export async function reserveInventory(
  orderId: string,
  items: Array<{ productId: string; quantity: number }>
): Promise {
  const ctx = Context.current();
  ctx.heartbeat('reserving inventory');

  const response = await axios.post('http://inventory-service/reserve', {
    orderId,
    items,
  });
  return response.data.reservationId;
}

export async function chargePayment(
  orderId: string,
  amount: number,
  paymentMethod: string
): Promise {
  const response = await axios.post('http://payment-service/charge', {
    orderId,
    amount,
    method: paymentMethod,
  });
  return response.data.transactionId;
}

export async function sendNotification(
  userId: string,
  message: string
): Promise {
  await axios.post('http://notification-service/send', {
    userId,
    message,
  });
}

// 补偿 Activity（Saga 回滚）
export async function releaseInventory(reservationId: string): Promise {
  await axios.post(`http://inventory-service/release/${reservationId}`);
}

export async function refundPayment(transactionId: string): Promise {
  await axios.post(`http://payment-service/refund/${transactionId}`);
}

2.3 定义 Workflow：Saga 编排

// src/workflows/order-saga.ts
import {
  defineQuery,
  defineSignal,
  setHandler,
  sleep,
  proxyActivities,
} from '@temporalio/workflow';

import type * as activities from '../activities';

const {
  reserveInventory,
  chargePayment,
  sendNotification,
  releaseInventory,
  refundPayment,
} = proxyActivities({
  startToCloseTimeout: '30 seconds',
  retry: {
    initialInterval: '1 second',
    maximumInterval: '30 seconds',
    backoffCoefficient: 2,
    maximumAttempts: 5,
    nonRetryableErrorTypes: ['InvalidOrderError', 'InsufficientStockError'],
  },
  heartbeatTimeout: '10 seconds',
});

export const getOrderStatus = defineQuery('getOrderStatus');
export const cancelOrder = defineSignal('cancelOrder');

export interface OrderInput {
  orderId: string;
  userId: string;
  items: Array<{ productId: string; quantity: number }>;
  totalAmount: number;
  paymentMethod: string;
}

export async function orderSagaWorkflow(input: OrderInput): Promise {
  let status = 'PROCESSING';
  let reservationId: string | undefined;
  let transactionId: string | undefined;

  setHandler(getOrderStatus, () => status);
  setHandler(cancelOrder, () => {
    if (status === 'PROCESSING') status = 'CANCELLED';
  });

  try {
    // Step 1: 预留库存
    status = 'RESERVING_INVENTORY';
    reservationId = await reserveInventory(input.orderId, input.items);

    // Step 2: 扣款
    status = 'CHARGING_PAYMENT';
    transactionId = await chargePayment(
      input.orderId,
      input.totalAmount,
      input.paymentMethod
    );

    // Step 3: 发送通知
    status = 'SENDING_NOTIFICATION';
    await sendNotification(
      input.userId,
      `Order ${input.orderId} confirmed! Total: $${input.totalAmount}`
    );

    status = 'COMPLETED';
    return `Order ${input.orderId} processed successfully`;

  } catch (error) {
    status = 'COMPENSATING';

    // Saga 补偿：按反向顺序执行补偿操作
    if (transactionId) {
      console.log(`Refunding payment ${transactionId}...`);
      await refundPayment(transactionId);
    }
    if (reservationId) {
      console.log(`Releasing inventory ${reservationId}...`);
      await releaseInventory(reservationId);
    }

    status = 'FAILED';
    throw error;
  }
}

三、生产级特性深度实战

3.1 定时器与延时调度

import { sleep, continueAsNew } from '@temporalio/workflow';

// 订单超时自动取消（30分钟未支付）
export async function orderWithTimeout(orderId: string): Promise {
  let paid = false;

  setHandler(markPaidSignal, () => { paid = true; });

  const timerDone = await Promise.race([
    sleep('30 minutes').then(() => true),
  ]);

  if (!paid) {
    await executeActivity('cancelOrder', { orderId });
    return false;
  }
  return true;
}

// 定时批处理工作流（每天凌晨执行）
export async function dailyReportWorkflow(): Promise {
  while (true) {
    await executeActivity('generateDailyReport', {});
    await sleep({ hours: 24 });
  }
}

// 超长运行工作流续新（防止事件历史过大）
export async function longRunningMonitoring(checkpoint: number): Promise {
  for (let i = 0; i < 1000>

import {
  executeChild,
  ChildWorkflowCancellationType,
} from '@temporalio/workflow';

// 审批链工作流
export async function orderApprovalWorkflow(
  orderId: string,
  amount: number
): Promise {
  // 并行执行多个审批
  const results = await Promise.all([
    executeChild('managerApproval', {
      args: [{ orderId, amount, level: 'L1' }],
      cancellationType: ChildWorkflowCancellationType.WAIT_CANCELLATION_COMPLETED,
    }),
    executeChild('riskCheckWorkflow', {
      args: [{ orderId, amount }],
    }),
  ]);

  const [approved, safe] = results;
  return approved && safe;
}

import { deprecatePatch, patched } from '@temporalio/workflow';

// 版本迁移示例：添加新步骤
export async function enhancedOrderSaga(input: OrderInput): Promise {
  if (patched('v2-add-fraud-check')) {
    const passed = await executeActivity('fraudDetection', { input });
    if (!passed) throw new Error('Fraud detected');
  }

  deprecatePatch('v1-no-fraud-check');

  return await commonOrderLogic(input);
}

# temporal-values.yaml
server:
  replicaCount: 3
  config:
    persistence:
      default:
        sql:
          host: temporal-postgresql
          port: 5432
          database: temporal
          user: temporal
          password: \${DB_PASSWORD}
      visibility:
        sql:
          host: temporal-postgresql
          port: 5432
          database: temporal_visibility
          user: temporal
          password: \${DB_PASSWORD}

frontend:
  replicaCount: 3
  resources:
    requests: { cpu: "500m", memory: "512Mi" }
    limits: { cpu: "2000m", memory: "2Gi" }

history:
  replicaCount: 5
  resources:
    requests: { cpu: "1000m", memory: "1Gi" }
    limits: { cpu: "4000m", memory: "4Gi" }

matching:
  replicaCount: 3

worker:
  replicaCount: 2

dynamicConfigValues:
  frontend.enableUpdateWorkflowExecution:
    - value: true
  history.persistenceMaxQPS:
    - value: 3000

# 创建隔离命名空间
temporal operator namespace create --retention 30 order-service
temporal operator namespace create --retention 7 payment-service
temporal operator namespace create --retention 1 log-service

# 设置权限策略
temporal operator namespace update order-service \
  --acl '[{"actions":["WRITE"],"users":["[email protected]"]}]'

# prometheus-scrape-config.yaml
scrape_configs:
  - job_name: 'temporal'
    kubernetes_sd_configs:
      - role: pod
    relabel_configs:
      - source_labels: [__meta_kubernetes_pod_label_app]
        regex: temporal
        action: keep
    metrics_path: /metrics
    scrape_interval: 15s

// 幂等性设计示例
export async function idempotentCharge(
  orderId: string,
  amount: number
): Promise {
  // 先检查是否已有成功记录
  const existing = await db.query(
    'SELECT * FROM payments WHERE order_id = ? AND status = ?',
    [orderId, 'SUCCESS']
  );
  if (existing.length > 0) return existing[0].transaction_id;

  // 首次执行，创建带幂等键的记录
  await db.query(
    'INSERT INTO payments (order_id, amount, status) VALUES (?, ?, ?)',
    [orderId, amount, 'PROCESSING']
  );
  // 执行实际扣款...
}