Temporal Activity取消的最佳实践
Temporal Activity取消的最佳实践
背景
在Temporal工作流开发中，我们遇到了一个典型的取消延迟问题：当Workflow发送Cancel请求后，Activity并未立即终止，而是继续执行了多个周期后才退出。以下是核心场景复现：
// Workflow
func MyWorkflow(ctx workflow.Context) error {
ao := workflow.ActivityOptions{
StartToCloseTimeout: 30 * time.Minute,
HeartbeatTimeout: 3 * time.Minute,
WaitForCancellation: true,
}
// ...忽略部分代码
// 添加一个定时器以在20秒后取消活动
selector.AddFuture(workflow.NewTimer(ctx, 20*time.Second), func(f workflow.Future) {
cancel()
logger.Info("ActivityToBeCanceled canceled")
})
// 执行活动
future := workflow.ExecuteActivity(activityCtx, a.ActivityToBeCanceled)
// ...忽略部分代码
selector.Select(ctx)
logger.Info("Workflow Execution complete.")
return nil
}
// Activity
type Activities struct{}
func (a *Activities) ActivityToBeCanceled(ctx context.Context) (string, error) {
logger := activity.GetLogger(ctx)
for {
select {
case <-time.After(10 * time.Second):
logger.Info("heartbeating...")
activity.RecordHeartbeat(ctx, "") //记录心跳
if ctx.Err() != nil {
logger.Info("context err is not nil", ctx.Err())
continue
}
// do soming...
case <-ctx.Done():
logger.Info("context is cancelled")
return "I am canceled by Done", nil
}
}
}整体流程是：工作流启动长时间运行的Activity，同时设置20秒后自动取消该Activity。Activity会响应取消信号并正常结束执行。参考官方demo
执行的日志如下（精简后）：
16:25:44 - Workflow - 工作流启动: "cancel workflow started"
16:25:44 - Workflow - 设置定时器: "NewTimer ... Duration 20s"
16:25:44 - Workflow - 执行活动: "ExecuteActivity ... ActivityType ActivityToBeCanceled"
16:25:44 - Activity - 活动开始: "activity started, to cancel the Workflow Execution..."
16:25:44 - Activity - 活动心跳: "heartbeating..."
16:25:44 - Activity - 活动工作: "do some work..."
16:25:47 - Activity - 工作完成: "work done"
16:26:00 - Activity - 活动心跳: "heartbeating..."
16:26:00 - Activity - 活动工作: "do some work..."
16:26:03 - Activity - 工作完成: "work done"
16:26:04 - Workflow - 取消请求: "RequestCancelActivity"
16:26:04 - Workflow - 活动取消: "ActivityToBeCanceled canceled"
16:26:04 - Workflow - 工作流完成并退出: "Workflow Execution complete."
16:26:14 - Activity - 活动心跳: "heartbeating..."
16:26:14 - Activity - 活动工作: "do some work..."
16:26:17 - Activity - 工作完成: "work done"
16:26:31 - Activity - 活动心跳: "heartbeating..."
16:26:31 - Activity - 活动工作: "do some work..."
16:26:34 - Activity - 工作完成: "work done"
16:26:44 - Activity - 心跳错误: "RecordActivityHeartbeat with error ... workflow execution already completed"
16:26:44 - Activity - 上下文取消，活动退出: "context is cancelled"异常现象：工作流在16:26:04发送取消请求，但Activity直到16:26:44（40秒后）才响应取消。期间Activity继续执行了两次完整的工作周期。Why？O.o
查阅资料
这里显示了查阅过程，想看结论请直接跳转问题根因分析
官方论坛
遇事不决先谷歌
先在Temporal官方论坛搜索相关问题：activity cancel delay
官方论坛可以解决95%的问题，上面的成员会很友好很耐心地回答你的问题。YYDS
关键：通过查阅资料发现，Temporal在Activity心跳上报机制上，做了限流
https://community.temporal.io/t/problems-cancelling-a-running-activity-from-parent-workflow/2169/9
Temporal老板说了几点很重要的信息：
- 心跳批处理优化：为了避免过多调用服务，心跳信号并不会立即发送，而是等到达到心跳间隔的80%才会发送到服务端。
- 加速取消通知：如果你希望加快活动收到取消/工作流完成通知的速度，应该将心跳间隔从默认值(例如1分钟)降低到更小的值。
- 取消后清理：如果需要在当前活动取消后执行清理工作，可以将ActivityOptions.WaitForCancellation设置为true，这样系统会通过阻塞活动的Future来等待活动取消完成。
这些信息对于理解Temporal中活动取消机制的工作原理以及如何优化取消响应速度非常重要。
上面代码中，我们的心跳间隔设置是3分钟：
// Workflow
func MyWorkflow(ctx workflow.Context) error {
ao := workflow.ActivityOptions{
StartToCloseTimeout: 30 * time.Minute,
HeartbeatTimeout: 3 * time.Minute,
WaitForCancellation: true,
}
ctx = workflow.WithActivityOptions(ctx, ao)
// 省略...
}上报心跳的时候会一并检查Activity是否被取消，也就是说，文中设置3分钟的心跳超时，心跳间隔会设置在2.4分钟（144秒）才会检测一次。
源码
在官方SDK中，我们可以找到Heartbeat的源码
// go.temporal.io › sdk@v1.26.0 › internal › internal_task_handlers.go
func (i *temporalInvoker) Heartbeat(ctx context.Context, details *commonpb.Payloads, skipBatching bool) error {
i.Lock()
defer i.Unlock()
if i.hbBatchEndTimer != nil && !skipBatching {
// If we have started batching window, keep track of last reported progress.
i.lastDetailsToReport = &details
return nil
}
isActivityCanceled, err := i.internalHeartBeat(ctx, details)
// If the activity is canceled, the activity can ignore the cancellation and do its work
// and complete. Our cancellation is co-operative, so we will try to heartbeat.
if (err == nil || isActivityCanceled) && !skipBatching {
// We have successfully sent heartbeat, start next batching window.
i.lastDetailsToReport = nil
// Create timer to fire before the threshold to report.
i.hbBatchEndTimer = time.NewTimer(i.heartbeatThrottleInterval)
go func() {
select {
case <-i.hbBatchEndTimer.C:
// We are close to deadline.
case <-i.workerStopChannel:
// Activity worker is close to stop. This does the same steps as batch timer ends.
case <-i.closeCh:
// We got closed.
return
}
// We close the batch and report the progress.
var detailsToReport **commonpb.Payloads
i.Lock()
detailsToReport = i.lastDetailsToReport
i.hbBatchEndTimer.Stop()
i.hbBatchEndTimer = nil
i.Unlock()
if detailsToReport != nil {
// TODO: there is a potential race condition here as the lock is released here and
// locked again in the Hearbeat() method. This possible that a heartbeat call from
// user activity grabs the lock first and calls internalHeartBeat before this
// batching goroutine, which means some activity progress will be lost.
_ = i.Heartbeat(ctx, *detailsToReport, false)
}
}()
}
return err
}Activity Heartbeat 是 Temporal 工作流中的一个关键机制，主要用于长时间运行的活动。根据代码分析：
-
心跳机制目的：
-
允许长时间运行的活动向 Temporal 服务器报告其进度
-
检测活动是否被取消
-
防止超时导致的活动失败
-
-
关键实现细节：
-
心跳批处理：通过
heartbeatThrottleInterval控制发送频率，避免过于频繁的心跳信号 -
最新进度跟踪：使用
lastDetailsToReport记录最新进度 -
协作式取消：即使活动被取消，系统也会尝试发送心跳
-
-
心跳限流处理流程：
-
如果批处理窗口开启（
skipBatching=false），只记录进度不立即发送 -
批处理窗口（
heartbeatThrottleInterval）结束或工作线程停止时，发送最新进度报告
-
问题根因分析
关键机制解析
通过查阅Temporal官方文档和社区讨论，结合源码分析，我们发现以下核心机制：
- 心跳批处理机制
- 心跳上报默认采用
80% HeartbeatTimeout的间隔进行批处理 - 示例中的
HeartbeatTimeout=3m，实际检测间隔为：3m * 0.8 = 144秒 - 活动只有在心跳时才会检测取消信号
- 心跳上报默认采用
- 服务端通知延迟
- 取消信号需要等待下一个心跳窗口才能传递到Activity
- 即使工作流已关闭，活动实例仍可能继续运行直到检测到取消
- SDK实现细节
// SDK内部处理逻辑（节选）
func (i *temporalInvoker) Heartbeat(ctx context.Context, ...) error {
if i.hbBatchEndTimer != nil { // 批处理窗口存在时
i.lastDetailsToReport = &details // 仅记录最新状态
return nil // 不立即发送心跳
}
// 实际发送心跳后启动新的批处理窗口
i.hbBatchEndTimer = time.NewTimer(i.heartbeatThrottleInterval)
}问题诊断
| 时间轴 | 事件 |
|---|---|
| 16:25:44 | 活动启动 |
| 16:26:04 | 工作流发送取消请求 |
| 16:26:14-16:26:44 | 活动继续执行两轮周期（10秒/轮） |
| 16:26:44 | 检测到取消信号 |
由于心跳间隔设置过大（3分钟），实际检测窗口为144秒，导致取消信号延迟传递。
优化心跳配置
// 推荐配置示例
ao := workflow.ActivityOptions{
HeartbeatTimeout: 10 * time.Second, // <= 建议10-30秒
StartToCloseTimeout: 30 * time.Minute,
WaitForCancellation: true, // 确保清理完成
}配置原则：
- 心跳超时 ≤ 活动单次操作耗时的2倍
- 对于需要快速响应的场景，设置
HeartbeatTimeout=10s（实际检测间隔8秒）
总结
问题本质：Temporal的心跳机制通过批处理优化服务端负载，导致取消信号检测存在固有延迟。当HeartbeatTimeout设置过大时，会显著放大这种延迟效应。
实践建议：根据业务场景设置合理的心跳间隔（建议小于1分钟），平衡响应速度与系统负载
Tips: 还可以等Temporal修复这个问题(^-^)
本博客所有文章除特别声明外，均采用 CC BY-NC-SA 4.0 许可协议。转载请注明来源 被咬了一口の包子°！
- 微信
- 支付宝
相关推荐
2025-10-29
Temporal Workflow 测试用例编写指南
Temporal Workflow 测试用例编写指南 引言 篇指南旨在为初次使用 Temporal 的开发者提供一个清晰、易懂的 Workflow 测试用例编写框架。Temporal 作为一个强大的工作流编排引擎，其测试框架也同样功能丰富。 为了循序渐进，本次我们先将重点放在 Workflow 的测试用例编写上，关于 Activity 的测试用例，我们将在下一篇文章中详细探讨。 Workflow 测试的两大核心 要成功运行一个 Workflow 测试用例，重点依赖于两个关键部分：测试环境初始化 和 Mock Activity。 1. 测试环境初始化 Temporal 提供了一个专用的测试框架，允许你在内存中执行 Workflow，而无需依赖真实的 Temporal 服务。这大大加快了测试速度并简化了环境配置。 一个典型的初始化过程如下： // 引入测试包 import ( "testing" "github.com/stretchr/testify/suite" "go.temporal.io/sdk/testsuite" ) // 定义测试套件 type MyWorkf...
2022-07-10
Reids快速入门
String 是什么？ String就是字符串，它是Redis中最基本的数据对象，最大为512MB。 基本操作 写操作 SET 语法：SET key value 功能：设置一个key的值为特定的value，成功则返回OK。 String：对象的创建或者更新都是该命令。 127.0.0.1:6379>SET str cat OK SETNX 语法：SETNX key value 功能：用于在指定的ky不存在时，为key设置指定的值，返回值0表示「key存在」不做操作，1表示设置成功。 如果对存在的Key，调用SETNX： 127.0.0.1:6379>SETNX str fish (integer)0 对不存在的key，调用SETNX： 127.0.0.1:6379>SETNX strmart fish (integer)1 DEL 语法：DEL key[key] 功能：删除对象，返回值为删除成功了几行。 127.0.0.1:6379>DEL str (integer)1 读操作 GET 语法：GET key 功能：查询某个key，存在就返回对应的value，如果不存在返回ni...
2023-02-24
SpringBoot快速入门
SpringBoot起步依赖 Starter SpringBoot中常见项目名称，定义了当前项目使用的所有项目坐标，以达到减少依赖配置的目的 parent 所有SpringBoot项目要继承的项目，定义了若干个坐标版本号（依赖管理，而非依赖），以达到减少依赖冲突的目的 spring-boot-starter-parent（2.5.0)与spring-boot-starter-parent（2.4.6)共计57处坐标版本不同 实际开发 使用任意坐标时，仅书写GAV中的G和A，V由SpringBoot提供 如发生坐标错误，再指定version(要小心版本冲突) SpringBoot程序启动 启动方式 SpringBoot的引导类是项目的入口，运行成main方法就可以启动项目 SpringBoot在创建项目时，采用jar的打包方式 更变起步依赖（starter） 使用maven依赖管理变更起步依赖项 Jetty比Tomcat更轻量级，可扩展性更强（相较于Tomcat)，谷歌应用引擎(GAE)已经全面切换为Jetty 基础配置 配置格式 SpringBoot提供...
2023-02-11
MyBatis入门
介绍 什么是MyBatis MyBatis是一款优秀的持久层（Dao）框架，用于简化JDBC开发 MyBatis本是Apache的一个开源项目iBatis，2010年这个项目由apache software foundation迁移到了google code，并且改名为MyBatis。2013年11月迁移到Github 官网：https://mybatis.org/mybatis-3/zh/index.html MyBatis 优点 如何写 Mapper代理开发 定义与SQL映射文件同名的Mapper接口，并且将Mapper接口和SQL映射文件放置在同一目录下 设置SQL映射文件的namespace属性为Mapper接口全限定名 在Mapper接口中定义方法，方法名就是SQL映射文件中sql语句的id，并保持参数类型和返回值类型一致 编码 通过SqlSession的getMapper方法获取Mapper接口的代理对象 调用对应方法完成sq的执行 细节：如果Mapper接口名称和SQL映射文件名称相同，并在同一目录下，则可以使用...
2023-07-13
Nginx过滤带有下划线('_')的请求头
在最近开发中遇到一个问题，记录一下。 发现问题 有一个http请求，需要携带一个请求头Header，当时定义的请求头使用的下划线模式，header_name这样。然后神奇的事情就发生了。 我自己在本机运行代码，在ApiFox上测试没问题，后端能正常接收并响应。 当部署到服务器后，使用域名或者ip访问接口路径，这时候后端就接收不到前端塞的请求头了。 排查问题 排查了好久，其中当尝试换成其他的请求头，发现只有携带下划线(_)才接收不到，换成 - 短横就可以了。 研究参考 搜了一下，发现原来是nginx的问题。ngixn默认不支持下划线的header，带下划线的header会被丢弃。 所以一开始自己ApiFox测试是因为直接请求的本地运行的代码，没有经过nginx，所以能正常接收。但后面都是使用域名经过nginx转发，导致带下划线的header丢失。 Nginx默认不支持带有下划线的Header字段，这是由于历史原因。在早期的互联网规范中，HTTP Header字段名中并不允许使用下划线。因此，为了遵守这些规范并防止可能的冲突，Nginx在处理Headers时默认会丢弃任何带有下划线...
2023-02-10
Spring入门
Spring快速入门 开发步骤 导入Spring开发的基本包坐标（Maven） 编写Dao接口和实现类 创建Spring核心配置文件 在Spring配置文件中配置UserDaoImpl 使用Spring的API获得Bean实例 applicationXml配置文件 Bean标签的属性 scope：指对象的作用范围 注： singleton：加载配置文件时创建bean对象 prototype：getBean时创建bean对象 init-method：指定类中的初始化方法名称 destroy-method：指定类中销毁方法名称 Bean实例化三种方式 无参构造方法实例化（常用） 工厂静态方法实例化 工厂实例方法实例化 Bean的依赖注入 依赖注入：它是Spring框架核心IOC的具体实现。 在编写程序时，通过控制反转，把对象的创建交给了 Spring，但是代码中不可能出现没有依赖的情况。 IOC解耕只是降低他们的依赖关系，但不会消除。例如：业务层仍会调用持久层的方法。 那这种业务层和持久层的依赖关系，在使用 Spring 之后，就让Spring 来维护...
评论