Skip to content

信号、查询与更新 ​

Temporal 提供三种机制让外部系统与运行中的工作流交互:Signal(信号)、Query(查询)和 Update(更新)。本文档详细介绍这三种机制的使用方法和最佳实践。


Signal(信号) ​

Signal 是外部向工作流发送消息的机制,用于异步通知工作流执行某些操作。

Signal 特点 ​

特性说明
异步发送发送后立即返回,不等待处理完成
不阻塞发送方发送方无需等待工作流响应
触发工作流逻辑工作流收到信号后执行相应逻辑
持久化信号事件被记录到事件历史
可多次发送同一工作流可接收多个同名信号

工作流接收 Signal ​

go
package workflow

import (
    "time"
    "go.temporal.io/sdk/workflow"
)

func OrderWorkflow(ctx workflow.Context, orderID string) error {
    logger := workflow.GetLogger(ctx)
    
    var orderStatus string = "created"
    
    // 设置查询处理器(用于查询当前状态)
    err := workflow.SetQueryHandler(ctx, "get-status", func() (string, error) {
        return orderStatus, nil
    })
    if err != nil {
        return err
    }
    
    // 监听支付信号
    paymentChan := workflow.GetSignalChannel(ctx, "payment-received")
    
    // 监听取消信号
    cancelChan := workflow.GetSignalChannel(ctx, "cancel-order")
    
    // 使用 Selector 处理多个信号
    for {
        selector := workflow.NewSelector(ctx)
        
        selector.AddReceive(paymentChan, func(c workflow.ReceiveChannel, more bool) {
            var payment PaymentInfo
            c.Receive(ctx, &payment)
            logger.Info("Received payment signal", "amount", payment.Amount)
            orderStatus = "paid"
        })
        
        selector.AddReceive(cancelChan, func(c workflow.ReceiveChannel, more bool) {
            var reason string
            c.Receive(ctx, &reason)
            logger.Info("Received cancel signal", "reason", reason)
            orderStatus = "cancelled"
        })
        
        selector.AddFuture(workflow.NewTimer(ctx, 24*time.Hour), func(f workflow.Future) {
            logger.Info("Order timeout")
            orderStatus = "timeout"
        })
        
        selector.Select(ctx)
        
        if orderStatus == "paid" {
            // 处理支付完成后的逻辑
            ao := workflow.ActivityOptions{
                StartToCloseTimeout: 5 * time.Minute,
            }
            ctx = workflow.WithActivityOptions(ctx, ao)
            
            err := workflow.ExecuteActivity(ctx, ProcessOrder, orderID).Get(ctx, nil)
            if err != nil {
                return err
            }
            orderStatus = "completed"
            return nil
        }
        
        if orderStatus == "cancelled" || orderStatus == "timeout" {
            workflow.ExecuteActivity(ctx, CancelOrder, orderID).Get(ctx, nil)
            return nil
        }
    }
}

type PaymentInfo struct {
    Amount   float64
    Currency string
    Method   string
}

外部发送 Signal ​

go
package client

import (
    "context"
    "go.temporal.io/sdk/client"
)

func SendPaymentSignal(c client.Client, workflowID string, payment PaymentInfo) error {
    // 发送信号到工作流
    err := c.SignalWorkflow(
        context.Background(),
        workflowID,
        "",  // runID 留空表示使用当前运行
        "payment-received",
        payment,
    )
    if err != nil {
        return err
    }
    
    return nil
}

func SendCancelSignal(c client.Client, workflowID string, reason string) error {
    err := c.SignalWorkflow(
        context.Background(),
        workflowID,
        "",
        "cancel-order",
        reason,
    )
    return err
}

Signal 批量处理 ​

go
func BatchSignalWorkflow(ctx workflow.Context) error {
    logger := workflow.GetLogger(ctx)
    
    var items []string
    
    signalChan := workflow.GetSignalChannel(ctx, "add-items")
    
    // 定期处理累积的信号
    for {
        selector := workflow.NewSelector(ctx)
        
        selector.AddReceive(signalChan, func(c workflow.ReceiveChannel, more bool) {
            var newItem string
            c.Receive(ctx, &newItem)
            items = append(items, newItem)
            logger.Info("Item added", "total", len(items))
        })
        
        // 每 5 秒处理一次
        selector.AddFuture(workflow.NewTimer(ctx, 5*time.Second), func(f workflow.Future) {
            if len(items) > 0 {
                logger.Info("Processing batch", "count", len(items))
                
                ao := workflow.ActivityOptions{
                    StartToCloseTimeout: time.Minute,
                }
                ctx = workflow.WithActivityOptions(ctx, ao)
                
                workflow.ExecuteActivity(ctx, ProcessBatch, items).Get(ctx, nil)
                items = nil  // 清空已处理的项
            }
        })
        
        selector.Select(ctx)
    }
}

Query(查询) ​

Query 是同步查询工作流状态的机制,不改变工作流状态,只返回当前信息。

Query 特点 ​

特性说明
同步调用发送后等待响应返回
不改变状态仅查询,不影响工作流执行
不记录历史Query 事件不记录到事件历史
必须快速响应Query 处理应立即返回
确定性要求Query 处理器必须具有确定性

设置 Query Handler ​

go
func OrderQueryWorkflow(ctx workflow.Context, orderID string) error {
    logger := workflow.GetLogger(ctx)
    
    // 工作流状态
    state := &OrderState{
        OrderID:     orderID,
        Status:      "created",
        TotalAmount: 0,
        Items:       []string{},
    }
    
    // 设置查询处理器:获取订单状态
    err := workflow.SetQueryHandler(ctx, "get-status", func() (*OrderState, error) {
        return state, nil
    })
    if err != nil {
        return err
    }
    
    // 设置查询处理器:获取订单历史
    var history []string
    err = workflow.SetQueryHandler(ctx, "get-history", func() ([]string, error) {
        return history, nil
    })
    if err != nil {
        return err
    }
    
    // 设置查询处理器:检查是否可取消
    err = workflow.SetQueryHandler(ctx, "can-cancel", func() (bool, error) {
        return state.Status == "created" || state.Status == "paid", nil
    })
    if err != nil {
        return err
    }
    
    // 工作流逻辑...
    history = append(history, "Order created")
    
    // 处理支付
    ao := workflow.ActivityOptions{
        StartToCloseTimeout: 5 * time.Minute,
    }
    ctx = workflow.WithActivityOptions(ctx, ao)
    
    workflow.ExecuteActivity(ctx, ProcessPayment, orderID).Get(ctx, nil)
    state.Status = "paid"
    history = append(history, "Payment processed")
    
    return nil
}

type OrderState struct {
    OrderID     string
    Status      string
    TotalAmount float64
    Items       []string
}

外部执行 Query ​

go
func QueryOrderStatus(c client.Client, workflowID string) (*OrderState, error) {
    // 查询工作流状态
    resp, err := c.QueryWorkflow(
        context.Background(),
        workflowID,
        "",  // runID
        "get-status",
    )
    if err != nil {
        return nil, err
    }
    
    var state OrderState
    err = resp.Decode(&state)
    if err != nil {
        return nil, err
    }
    
    return &state, nil
}

func QueryOrderHistory(c client.Client, workflowID string) ([]string, error) {
    resp, err := c.QueryWorkflow(
        context.Background(),
        workflowID,
        "",
        "get-history",
    )
    if err != nil {
        return nil, err
    }
    
    var history []string
    err = resp.Decode(&history)
    return history, err
}

func QueryCanCancel(c client.Client, workflowID string) (bool, error) {
    resp, err := c.QueryWorkflow(
        context.Background(),
        workflowID,
        "",
        "can-cancel",
    )
    if err != nil {
        return false, err
    }
    
    var canCancel bool
    err = resp.Decode(&canCancel)
    return canCancel, err
}

Query 使用场景 ​

go
// 场景 1: 监控工作流进度
func ProgressTrackingWorkflow(ctx workflow.Context, taskID string) error {
    var progress int = 0
    var currentStep string = "initializing"
    
    workflow.SetQueryHandler(ctx, "get-progress", func() (map[string]interface{}, error) {
        return map[string]interface{}{
            "progress":    progress,
            "currentStep": currentStep,
            "taskID":      taskID,
        }, nil
    })
    
    ao := workflow.ActivityOptions{
        StartToCloseTimeout: 10 * time.Minute,
    }
    ctx = workflow.WithActivityOptions(ctx, ao)
    
    steps := []string{"download", "parse", "transform", "validate", "save"}
    
    for i, step := range steps {
        currentStep = step
        workflow.ExecuteActivity(ctx, ProcessStep, step).Get(ctx, nil)
        progress = (i + 1) * 100 / len(steps)
    }
    
    return nil
}

// 场景 2: 获取审批状态
func ApprovalQueryWorkflow(ctx workflow.Context, requestID string) error {
    type ApprovalStatus struct {
        RequestID    string
        Status       string
        CurrentLevel int
        Approver     string
        Comments     []string
    }
    
    status := &ApprovalStatus{
        RequestID:    requestID,
        Status:       "pending",
        CurrentLevel: 0,
        Approver:     "",
        Comments:     []string{},
    }
    
    workflow.SetQueryHandler(ctx, "get-approval-status", func() (*ApprovalStatus, error) {
        return status, nil
    })
    
    // 审批流程逻辑...
    for i := 1; i <= 3; i++ {
        status.CurrentLevel = i
        status.Approver = fmt.Sprintf("approver-%d", i)
        
        signalChan := workflow.GetSignalChannel(ctx, "approval-decision")
        var decision ApprovalDecision
        signalChan.Receive(ctx, &decision)
        
        status.Comments = append(status.Comments, decision.Comment)
        
        if decision.Approved {
            status.Status = fmt.Sprintf("approved-level-%d", i)
        } else {
            status.Status = "rejected"
            return nil
        }
    }
    
    status.Status = "approved"
    return nil
}

Update(更新) ​

Update 是 Temporal 新引入的特性,允许外部同步更新工作流状态。

Update 特点 ​

特性说明
同步调用等待更新完成后返回
可改变状态可以修改工作流状态
可返回结果返回更新后的状态
有验证阶段支持验证请求是否可接受
记录历史Update 事件记录到事件历史

设置 Update Handler ​

go
func OrderUpdateWorkflow(ctx workflow.Context, orderID string) error {
    logger := workflow.GetLogger(ctx)
    
    state := &OrderState{
        OrderID:     orderID,
        Status:      "created",
        Items:       []OrderItem{},
        TotalAmount: 0,
    }
    
    // 设置 Query Handler
    workflow.SetQueryHandler(ctx, "get-state", func() (*OrderState, error) {
        return state, nil
    })
    
    // 设置 Update Handler:添加商品
    workflow.SetUpdateHandler(ctx, "add-item", func(ctx workflow.Context, item OrderItem) (*OrderState, error) {
        // 验证:只能在使用 created 状态时添加
        if state.Status != "created" {
            return nil, fmt.Errorf("cannot add items after order is %s", state.Status)
        }
        
        state.Items = append(state.Items, item)
        state.TotalAmount += item.Price * item.Quantity
        logger.Info("Item added", "productID", item.ProductID, "total", state.TotalAmount)
        
        return state, nil
    })
    
    // 设置 Update Handler:移除商品
    workflow.SetUpdateHandler(ctx, "remove-item", func(ctx workflow.Context, productID string) (*OrderState, error) {
        if state.Status != "created" {
            return nil, fmt.Errorf("cannot remove items after order is %s", state.Status)
        }
        
        for i, item := range state.Items {
            if item.ProductID == productID {
                state.TotalAmount -= item.Price * item.Quantity
                state.Items = append(state.Items[:i], state.Items[i+1:]...)
                logger.Info("Item removed", "productID", productID)
                break
            }
        }
        
        return state, nil
    })
    
    // 设置 Update Handler:更新地址
    workflow.SetUpdateHandler(ctx, "update-address", func(ctx workflow.Context, address ShippingAddress) (*OrderState, error) {
        state.ShippingAddress = address
        logger.Info("Address updated")
        return state, nil
    })
    
    // 设置 Update Validator
    workflow.SetUpdateHandlerWithOptions(
        ctx,
        "add-item",
        func(ctx workflow.Context, item OrderItem) (*OrderState, error) {
            state.Items = append(state.Items, item)
            state.TotalAmount += item.Price * item.Quantity
            return state, nil
        },
        workflow.UpdateHandlerOptions{
            Validator: func(item OrderItem) error {
                // 验证商品数量
                if item.Quantity <= 0 {
                    return fmt.Errorf("quantity must be positive")
                }
                if item.Price <= 0 {
                    return fmt.Errorf("price must be positive")
                }
                // 验证订单状态
                if state.Status != "created" {
                    return fmt.Errorf("order already %s", state.Status)
                }
                return nil
            },
        },
    )
    
    // 等待提交信号
    signalChan := workflow.GetSignalChannel(ctx, "submit-order")
    signalChan.Receive(ctx, nil)
    
    state.Status = "submitted"
    
    // 处理订单...
    ao := workflow.ActivityOptions{
        StartToCloseTimeout: 5 * time.Minute,
    }
    ctx = workflow.WithActivityOptions(ctx, ao)
    
    workflow.ExecuteActivity(ctx, ProcessOrder, state).Get(ctx, nil)
    state.Status = "completed"
    
    return nil
}

type OrderItem struct {
    ProductID string
    Quantity  int
    Price     float64
}

type ShippingAddress struct {
    Street  string
    City    string
    Country string
    ZipCode string
}

外部执行 Update ​

go
func UpdateOrderAddItem(c client.Client, workflowID string, item OrderItem) (*OrderState, error) {
    // 执行更新
    updateResp, err := c.UpdateWorkflow(
        context.Background(),
        workflowID,
        "",  // runID
        "add-item",
        client.UpdateWorkflowOptions{
            Args: []interface{}{item},
        },
    )
    if err != nil {
        return nil, err
    }
    
    // 等待更新完成并获取结果
    var state OrderState
    err = updateResp.Get(context.Background(), &state)
    if err != nil {
        return nil, err
    }
    
    return &state, nil
}

func UpdateOrderRemoveItem(c client.Client, workflowID string, productID string) (*OrderState, error) {
    updateResp, err := c.UpdateWorkflow(
        context.Background(),
        workflowID,
        "",
        "remove-item",
        client.UpdateWorkflowOptions{
            Args: []interface{}{productID},
        },
    )
    if err != nil {
        return nil, err
    }
    
    var state OrderState
    err = updateResp.Get(context.Background(), &state)
    return &state, err
}

Signal、Query、Update 对比 ​

特性SignalQueryUpdate
调用方式异步同步同步
等待响应不等待等待响应等待完成
改变状态可以(通过触发逻辑)不能可以
记录历史记录不记录记录
使用场景外部通知、异步触发状态查询、监控状态更新、同步操作
返回值无有有
可验证性无验证无验证支持验证

使用场景对比 ​

场景推荐使用原因
用户取消订单Signal用户操作异步,不阻塞用户
支付完成通知Signal支付系统异步通知
查询订单状态Query只读操作,快速响应
监控工作流进度Query不干扰工作流执行
添加购物车商品Update需要同步确认添加成功
更新配送地址Update要验证并立即生效
审批决策Signal 或 Update取决于是否需要立即确认

综合示例 ​

完整的订单管理工作流 ​

go
package workflow

import (
    "fmt"
    "time"
    "go.temporal.io/sdk/workflow"
)

type OrderState struct {
    OrderID         string
    Status          string
    Items           []OrderItem
    TotalAmount     float64
    ShippingAddress ShippingAddress
    PaymentMethod   string
}

func ComprehensiveOrderWorkflow(ctx workflow.Context, orderID string) error {
    logger := workflow.GetLogger(ctx)
    
    state := &OrderState{
        OrderID: orderID,
        Status:  "draft",
        Items:   []OrderItem{},
    }
    
    // ========== Query Handlers ==========
    
    workflow.SetQueryHandler(ctx, "get-state", func() (*OrderState, error) {
        return state, nil
    })
    
    workflow.SetQueryHandler(ctx, "get-items", func() ([]OrderItem, error) {
        return state.Items, nil
    })
    
    workflow.SetQueryHandler(ctx, "get-total", func() (float64, error) {
        return state.TotalAmount, nil
    })
    
    // ========== Update Handlers ==========
    
    workflow.SetUpdateHandlerWithOptions(
        ctx,
        "add-item",
        func(ctx workflow.Context, item OrderItem) (*OrderState, error) {
            state.Items = append(state.Items, item)
            state.TotalAmount += item.Price * item.Quantity
            return state, nil
        },
        workflow.UpdateHandlerOptions{
            Validator: func(item OrderItem) error {
                if state.Status != "draft" {
                    return fmt.Errorf("order is %s, cannot modify", state.Status)
                }
                if item.Quantity <= 0 {
                    return fmt.Errorf("invalid quantity")
                }
                return nil
            },
        },
    )
    
    workflow.SetUpdateHandlerWithOptions(
        ctx,
        "remove-item",
        func(ctx workflow.Context, productID string) (*OrderState, error) {
            for i, item := range state.Items {
                if item.ProductID == productID {
                    state.TotalAmount -= item.Price * item.Quantity
                    state.Items = append(state.Items[:i], state.Items[i+1:]...)
                    return state, nil
                }
            }
            return nil, fmt.Errorf("item not found")
        },
        workflow.UpdateHandlerOptions{
            Validator: func(productID string) error {
                if state.Status != "draft" {
                    return fmt.Errorf("order is %s, cannot modify", state.Status)
                }
                return nil
            },
        },
    )
    
    workflow.SetUpdateHandler(ctx, "set-address", func(ctx workflow.Context, address ShippingAddress) error {
        state.ShippingAddress = address
        return nil
    })
    
    workflow.SetUpdateHandler(ctx, "set-payment-method", func(ctx workflow.Context, method string) error {
        state.PaymentMethod = method
        return nil
    })
    
    // ========== Signal Channels ==========
    
    submitChan := workflow.GetSignalChannel(ctx, "submit")
    cancelChan := workflow.GetSignalChannel(ctx, "cancel")
    paymentChan := workflow.GetSignalChannel(ctx, "payment-done")
    shippedChan := workflow.GetSignalChannel(ctx, "shipped")
    
    // ========== 状态循环 ==========
    
    for {
        selector := workflow.NewSelector(ctx)
        
        selector.AddReceive(submitChan, func(c workflow.ReceiveChannel, more bool) {
            if state.Status == "draft" && len(state.Items) > 0 {
                state.Status = "submitted"
                logger.Info("Order submitted", "total", state.TotalAmount)
            }
        })
        
        selector.AddReceive(cancelChan, func(c workflow.ReceiveChannel, more bool) {
            if state.Status == "draft" || state.Status == "submitted" {
                state.Status = "cancelled"
                logger.Info("Order cancelled")
            }
        })
        
        selector.AddReceive(paymentChan, func(c workflow.ReceiveChannel, more bool) {
            if state.Status == "submitted" {
                state.Status = "paid"
                logger.Info("Payment completed")
            }
        })
        
        selector.AddReceive(shippedChan, func(c workflow.ReceiveChannel, more bool) {
            if state.Status == "paid" {
                state.Status = "shipped"
                logger.Info("Order shipped")
            }
        })
        
        selector.AddFuture(workflow.NewTimer(ctx, 30*time.Minute), func(f workflow.Future) {
            if state.Status == "draft" {
                state.Status = "expired"
                logger.Info("Order expired")
            }
        })
        
        selector.Select(ctx)
        
        switch state.Status {
        case "submitted":
            ao := workflow.ActivityOptions{
                StartToCloseTimeout: 5 * time.Minute,
            }
            ctx = workflow.WithActivityOptions(ctx, ao)
            workflow.ExecuteActivity(ctx, InitiatePayment, state).Get(ctx, nil)
            
        case "paid":
            ao := workflow.ActivityOptions{
                StartToCloseTimeout: 10 * time.Minute,
            }
            ctx = workflow.WithActivityOptions(ctx, ao)
            workflow.ExecuteActivity(ctx, ProcessShipment, state).Get(ctx, nil)
            
        case "shipped":
            ao := workflow.ActivityOptions{
                StartToCloseTimeout: time.Minute,
            }
            ctx = workflow.WithActivityOptions(ctx, ao)
            workflow.ExecuteActivity(ctx, SendConfirmation, state).Get(ctx, nil)
            state.Status = "completed"
            return nil
            
        case "cancelled", "expired":
            return nil
        }
    }
}

最佳实践 ​

Signal 最佳实践 ​

go
// 1. 使用明确的信号名称
signalChan := workflow.GetSignalChannel(ctx, "payment-completed")  // 好
signalChan := workflow.GetSignalChannel(ctx, "sig1")              // 不推荐

// 2. 处理信号时验证状态
func OrderWorkflow(ctx workflow.Context) error {
    var status string = "pending"
    cancelChan := workflow.GetSignalChannel(ctx, "cancel")
    
    for {
        selector := workflow.NewSelector(ctx)
        
        selector.AddReceive(cancelChan, func(c workflow.ReceiveChannel, more bool) {
            var reason string
            c.Receive(ctx, &reason)
            
            // 验证是否可以取消
            if status == "pending" || status == "processing" {
                status = "cancelled"
                workflow.GetLogger(ctx).Info("Order cancelled", "reason", reason)
            } else {
                workflow.GetLogger(ctx).Warn("Cannot cancel order", "status", status)
            }
        })
        
        // ...
        selector.Select(ctx)
    }
}

// 3. 处理信号缓冲
func BufferedSignalWorkflow(ctx workflow.Context) error {
    signalChan := workflow.GetSignalChannel(ctx, "events")
    
    // 批量接收信号
    for {
        var events []Event
        
        // 收集一段时间内的所有信号
        timer := workflow.NewTimer(ctx, 5*time.Second)
        
        selector := workflow.NewSelector(ctx)
        selector.AddReceive(signalChan, func(c workflow.ReceiveChannel, more bool) {
            var event Event
            c.Receive(ctx, &event)
            events = append(events, event)
        })
        selector.AddFuture(timer, func(f workflow.Future) {})
        
        // 循环接收直到超时
        for selector.HasPending() {
            selector.Select(ctx)
            if len(events) >= 100 {  // 达到批量上限
                break
            }
        }
        
        if len(events) > 0 {
            // 处理批量事件
            processEvents(ctx, events)
        }
    }
}

Query 最佳实践 ​

go
// 1. Query Handler 必须快速返回
workflow.SetQueryHandler(ctx, "get-status", func() (string, error) {
    return status, nil  // 简单返回,无复杂计算
})

// 2. 不在 Query 中执行 Activity
workflow.SetQueryHandler(ctx, "bad-query", func() (Result, error) {
    var result Result
    err := workflow.ExecuteActivity(ctx, SomeActivity).Get(ctx, &result)  // 错误!
    return result, err
})

// 3. Query 返回值应包含必要信息
workflow.SetQueryHandler(ctx, "get-order-details", func() (OrderDetails, error) {
    return OrderDetails{
        OrderID:     state.OrderID,
        Status:      state.Status,
        TotalAmount: state.TotalAmount,
        Items:       state.Items,
        CreatedAt:   workflow.Now(ctx),
    }, nil
})

Update 最佳实践 ​

go
// 1. 使用 Validator 验证请求
workflow.SetUpdateHandlerWithOptions(
    ctx,
    "update-quantity",
    func(ctx workflow.Context, productID string, quantity int) error {
        for i, item := range state.Items {
            if item.ProductID == productID {
                state.Items[i].Quantity = quantity
                recalculateTotal(state)
                return nil
            }
        }
        return fmt.Errorf("product not found")
    },
    workflow.UpdateHandlerOptions{
        Validator: func(productID string, quantity int) error {
            if quantity <= 0 || quantity > 100 {
                return fmt.Errorf("invalid quantity: %d", quantity)
            }
            if state.Status != "draft" {
                return fmt.Errorf("order already submitted")
            }
            return nil
        },
    },
)

// 2. Update 应返回更新后的状态
workflow.SetUpdateHandler(ctx, "add-item", func(ctx workflow.Context, item Item) (*State, error) {
    state.Items = append(state.Items, item)
    return state, nil  // 返回更新后的状态
})

相关资源 ​