Temporal Go工作流：如何等待同一通道中每个待处理元素的信号？

问题描述 #

工作流中有一个切片，需要等待切片中每个处于Pending状态的元素都通过同一个信号通道收到对应的信号。现有代码无法正确等待所有消息接收完成。

用户代码 #

selector := workflow.NewSelector(ctx)
notificationSignalChan := workflow.GetSignalChannel(ctx, "my-channel")

for i := 0; i < len(container.Items); i++ {
    if container.Items[i].Status != status.Pending {
        continue
    }

    var expectedNotification events.Notification

    selector.AddReceive(notificationSignalChan, func(c workflow.ReceiveChannel, more bool) {
        // So it has to be explicitly consumed here
        c.Receive(ctx, &expectedNotification)
        idx := slices.IndexFunc(container.Items, func(item *model.Item) bool {
            return item.ID == notification.ItemID
        })
        
        recordedAt := workflow.Now(ctx)
        container.Items[idx].Status = status.Processed
        err = workflow.ExecuteActivity(ctx, activities.OnProcessed, container.Items[idx]).Get(ctx, nil)
        
        if err != nil {
            panic(err)
        }
    })
}

for i := 0; i < len(container.Items); i++ {
    if container.Items[i].Status != status.Pending {
        continue
    }

    selector.Select(ctx)
}

代码问题点 #

1. 循环变量捕获错误：循环中声明的expectedNotification被所有闭包共享，每次信号接收都会覆盖同一个变量，导致信号数据混乱，无法正确匹配到对应的元素。
2. 变量名笔误：处理函数中使用未定义的notification.ItemID，实际应为expectedNotification.ItemID，这会导致编译错误或逻辑异常。
3. Selector滥用：多次给同一个信号通道添加接收处理，Selector的Select调用只会触发其中一个处理函数，但由于闭包变量问题，无法保证每个Pending元素都被正确处理。

修复方案 #

方案一：直接循环接收指定次数信号 #

这种方式更简洁直接，先统计需要处理的Pending元素数量，然后循环接收对应次数的信号，逐个处理：

notificationSignalChan := workflow.GetSignalChannel(ctx, "my-channel")

// 统计需要处理的Pending元素数量
pendingCount := 0
for _, item := range container.Items {
    if item.Status == status.Pending {
        pendingCount++
    }
}

// 循环接收信号，处理每个Pending元素
for i := 0; i < pendingCount; i++ {
    var notification events.Notification
    // 接收信号
    if !notificationSignalChan.Receive(ctx, &notification) {
        // 通道关闭，处理异常
        panic("signal channel closed before all notifications received")
    }

    // 找到对应的元素
    idx := slices.IndexFunc(container.Items, func(item *model.Item) bool {
        return item.ID == notification.ItemID
    })
    if idx == -1 {
        // 信号对应元素不存在，跳过或记录日志
        continue
    }

    // 更新状态并执行活动
    container.Items[idx].Status = status.Processed
    recordedAt := workflow.Now(ctx)
    err := workflow.ExecuteActivity(ctx, activities.OnProcessed, container.Items[idx]).Get(ctx, nil)
    if err != nil {
        panic(err)
    }
}

方案二：正确使用Selector（修复闭包变量问题） #

如果一定要用Selector，需要修复闭包变量捕获问题，确保每个处理函数使用独立的变量：

selector := workflow.NewSelector(ctx)
notificationSignalChan := workflow.GetSignalChannel(ctx, "my-channel")

pendingCount := 0
for i := 0; i < len(container.Items); i++ {
    if container.Items[i].Status != status.Pending {
        continue
    }
    pendingCount++

    // 创建局部变量，避免闭包捕获循环变量
    var localNotification events.Notification
    selector.AddReceive(notificationSignalChan, func(c workflow.ReceiveChannel, more bool) {
        c.Receive(ctx, &localNotification)
        idx := slices.IndexFunc(container.Items, func(item *model.Item) bool {
            return item.ID == localNotification.ItemID
        })
        if idx == -1 {
            return
        }

        container.Items[idx].Status = status.Processed
        recordedAt := workflow.Now(ctx)
        err := workflow.ExecuteActivity(ctx, activities.OnProcessed, container.Items[idx]).Get(ctx, nil)
        if err != nil {
            panic(err)
        }
    })
}

// 等待所有Pending元素的信号处理完成
for i := 0; i < pendingCount; i++ {
    selector.Select(ctx)
}

关键说明 #

• 避免在循环闭包中直接使用循环变量，必须创建局部变量来保存当前迭代的状态，防止变量被后续迭代覆盖。
• 明确需要接收的信号数量，确保循环次数与Pending元素数量一致，保证所有元素都被处理。

内容的提问来源于stack exchange，提问作者Stray MCC