Temporal 是一个开源的分布式工作流引擎用于构建高可靠、可伸缩、可观察的业务逻辑如订单处理、支付对账、微服务编排等。temporalio是 Temporal 官方提供的 Python SDK客户端库用于在 Python 应用中定义工作流Workflow、活动Activity、定时器、信号Signal、查询Query等并与 Temporal Server通过 gRPC 通信交互。核心概念简述Workflow定义业务逻辑的“蓝图”以确定性方式编写不可依赖非确定性操作如time.time()、随机数等由 Temporal Server 持久化执行状态。Activity执行实际外部操作如调用 API、读写数据库、发送邮件是 Workflow 中的“工作单元”具备重试、超时、心跳等容错能力。Worker运行 Workflow 和 Activity 的进程需注册对应的 Workflow 和 Activity 实现。Clienttemporalio.client.Client用于启动 Workflow、发送 Signal、查询状态、获取结果等。Task QueueWorker 与 Client 之间的解耦桥梁Workflow 和 Activity 任务通过队列分发。基础使用示例启动一个简单 Workflowimportasynciofromtemporalioimportworkflow,activityfromtemporalio.clientimportClientfromtemporalio.workerimportWorker# 定义 Activityactivity.defnasyncdefsay_hello(name:str)-str:returnfHello,{name}!# 定义 Workflowworkflow.defnclassGreetingWorkflow:workflow.runasyncdefrun(self,name:str)-str:returnawaitworkflow.execute_activity(say_hello,name,start_to_close_timeouttimedelta(seconds10))# 启动 Worker需先运行 Temporal Serverasyncdefmain():clientawaitClient.connect(localhost:7233)workerWorker(client,task_queuehello-task-queue,workflows[GreetingWorkflow],activities[say_hello])awaitworker.run()if__name____main__:asyncio.run(main())注意需提前部署 Temporal Server可通过 Docker 快速启动docker run -p 7233:7233 -p 8233:8233 temporalio/auto-setup并确保网络可达。temporalioSDK 支持异步/await、类型提示、重试策略、可观测性OpenTelemetry 集成、版本升级变更工作流逻辑时保证向后兼容等高级特性。在 Temporal 中实现工作流的版本升级与向后兼容核心在于避免破坏工作流的确定性determinism并利用 Temporal 提供的版本控制原语如workflow.getVersion、workflow.skip、workflow.continueAsNew等安全演进逻辑。Temporal 不允许直接修改已部署的工作流代码尤其是正在运行的实例但支持渐进式升级——即新旧版本共存历史实例按原逻辑继续执行新启动实例使用新版逻辑。以下是关键实践策略✅ 1. 使用workflow.getVersion()进行语义化版本分支这是最推荐、最安全的方式。它基于 Temporal Server 内部的“变更ID”change ID做兼容性标记确保相同 change ID 在同一工作流类型下永不变更。fromtemporalioimportworkflowworkflow.defnclassMyWorkflow:workflow.runasyncdefrun(self)-str:# 定义一个唯一且永久不变的 change ID如 add-logging-stepvawaitworkflow.getVersion(add-logging-step,WorkflowVersion.V1,WorkflowVersion.V2)ifvWorkflowVersion.V1:# 旧逻辑resultawaitworkflow.execute_activity(old_activity)elifvWorkflowVersion.V2:# 新逻辑例如增加日志、新增步骤awaitworkflow.execute_activity(log_step)resultawaitworkflow.execute_activity(new_activity)returnresult⚠️ 注意WorkflowVersion.V1/V2是枚举常量int由 SDK 提供change_id字符串必须全局唯一且永不修改否则导致非确定性错误。✅ 2. 避免破坏性变更严禁以下操作会导致工作流重放失败Non-deterministic error使历史实例卡死修改工作流函数签名参数名/顺序/类型变更除非用dataclass__post_init__等兼容方式删除或重命名已注册的 Activity 名称需保留旧 Activity 实现或通过activity_method显式指定在工作流中引入非确定性操作如random.random()、time.time()、未 mock 的全局状态修改workflow.defn类名或模块路径影响历史工作流反序列化✅ 正确做法用workflow.method/activity.method显式绑定方法名解耦实现与注册名所有外部依赖通过 Activity 调用工作流仅含纯逻辑使用workflow.info().workflow_id或workflow.get_current_time()确定性时间替代time.time()。✅ 3. 使用continueAsNew主动切换版本适用于重大重构当逻辑差异过大、无法用getVersion分支覆盖时可在工作流中主动调用workflow.continue_as_new()启动新实例并传递必要状态ifshould_upgrade:# 将当前状态序列化传给新版本awaitworkflow.continue_as_new(args[current_state],memo{upgraded_from:v1.2},# 可指定新任务队列、执行超时等)⚠️ 注意continueAsNew会终止当前执行、创建全新工作流实例新 run_id因此需确保状态可完整迁移且下游系统能容忍“中断-重启”。✅ 4. 其他辅助机制Memo Search Attributes升级后可通过memo记录版本信息或用search_attributes标记实例所属版本便于可观测性与批量治理Worker 版本灰度通过不同task_queue如my-queue-v1,my-queue-v2部署新旧 Worker配合 Client 启动时指定队列实现流量切分Schema Evolution 工具对args/result类型变更建议使用pydantic.BaseModel或dataclasses__post_init__提供默认值/字段迁移逻辑。 总结原则“所有已启动的工作流实例必须永远能按原始代码重放成功”—— 这是 Temporal 确定性模型的铁律。版本升级不是“替换”而是“共存演进”。