3.基于 Temporal 的 Couchbase 动态 SQL 执行场景

在使用 Temporal 和 Go 语言 调用 Couchbase 执行 SQL 脚本时，可以通过动态参数传递到 SQL 脚本中，以下是完整实现的指南：

1. Temporal Workflow 的参数传递

Temporal 的 Workflow 支持接收动态参数，将这些参数传递给执行 SQL 的 Activity。在 Go 中可以通过结构体或键值对传递参数。

2. Couchbase N1QL 示例

Couchbase 支持通过 REST API 或 SDK 执行 N1QL 查询，并且可以使用参数化查询来防止 SQL 注入。

3. 工作流和 Activity 的结构

以下是实现动态参数化执行 Couchbase SQL 的完整示例：

Workflow 定义

package workflowsimport ("context""time""go.temporal.io/sdk/workflow"
)// QueryParameters 定义 SQL 参数结构
type QueryParameters struct {StartTime string `json:"start_time"`EndTime   string `json:"end_time"`BrandID   int    `json:"brand_id"`
}// Workflow 定义
func SQLWorkflow(ctx workflow.Context, params QueryParameters) error {ao := workflow.ActivityOptions{StartToCloseTimeout: time.Minute,}ctx = workflow.WithActivityOptions(ctx, ao)// 调用 Activity 执行 SQLerr := workflow.ExecuteActivity(ctx, ExecuteSQLActivity, params).Get(ctx, nil)if err != nil {return err}return nil
}

Activity 定义

package activitiesimport ("context""encoding/json""fmt""log""github.com/couchbase/gocb/v2" // Couchbase Go SDK
)// QueryParameters 定义 SQL 参数结构
type QueryParameters struct {StartTime string `json:"start_time"`EndTime   string `json:"end_time"`BrandID   int    `json:"brand_id"`
}// ExecuteSQLActivity 用于执行 Couchbase 的 SQL 脚本
func ExecuteSQLActivity(ctx context.Context, params QueryParameters) error {// 初始化 Couchbase 集群连接cluster, err := gocb.Connect("couchbase://localhost", gocb.ClusterOptions{Username: "username",Password: "password",})if err != nil {return fmt.Errorf("failed to connect to Couchbase: %w", err)}defer cluster.Close()// 定义 SQL 查询query := `SELECTpStatDate,brandId,storeId,SUM(incomeAmount) AS incomeAmt,SUM(refundAmount) AS refundAmt,SUM(actualAmount) AS actualAmtFROM ` + "`origin`.`dwd_meal_order`.`fact_batch_payment_summary`" + `WHERE modifyTime >= $startTime AND modifyTime < $endTime AND brandId = $brandIdGROUP BY pStatDate, brandId, storeId`// 执行 N1QL 查询，绑定参数rows, err := cluster.Query(query, &gocb.QueryOptions{NamedParameters: map[string]interface{}{"startTime": params.StartTime,"endTime":   params.EndTime,"brandId":   params.BrandID,},})if err != nil {return fmt.Errorf("failed to execute query: %w", err)}defer rows.Close()// 解析查询结果for rows.Next() {var result map[string]interface{}err := rows.Row(&result)if err != nil {log.Printf("failed to read row: %v", err)continue}// 输出结果jsonResult, _ := json.Marshal(result)log.Printf("Result: %s", jsonResult)}if err := rows.Err(); err != nil {return fmt.Errorf("query execution error: %w", err)}return nil
}

启动 Workflow

package mainimport ("context""log""go.temporal.io/sdk/client""go.temporal.io/sdk/worker""your_project/activities""your_project/workflows"
)func main() {// 创建 Temporal 客户端c, err := client.Dial(client.Options{})if err != nil {log.Fatalf("unable to create Temporal client: %v", err)}defer c.Close()// 创建 Workerw := worker.New(c, "sql-task-queue", worker.Options{})w.RegisterWorkflow(workflows.SQLWorkflow)w.RegisterActivity(activities.ExecuteSQLActivity)// 启动 Workererr = w.Run(worker.InterruptCh())if err != nil {log.Fatalf("unable to start Worker: %v", err)}// 启动 Workflowparams := workflows.QueryParameters{StartTime: "2024-10-01T00:00:00Z",EndTime:   "2024-10-31T23:59:59Z",BrandID:   123,}workflowOptions := client.StartWorkflowOptions{ID:        "sql-workflow-id",TaskQueue: "sql-task-queue",}we, err := c.ExecuteWorkflow(context.Background(), workflowOptions, workflows.SQLWorkflow, params)if err != nil {log.Fatalf("unable to execute Workflow: %v", err)}log.Printf("Workflow started: %s", we.GetID())
}

关键点解析

参数化 SQL 查询
- 使用 Couchbase N1QL 的命名参数（$paramName）传递动态参数。
- 在 Activity 中通过 NamedParameters 绑定参数值。
动态 Workflow 参数传递
- 在 Workflow 中定义参数结构（如 QueryParameters），通过 Temporal SDK 将参数传递到 Activity。
Activity 执行 SQL
- Activity 中初始化 Couchbase 连接，执行 SQL 查询，并处理返回结果。
动态配置时间范围和条件
- StartTime 和 EndTime 是动态参数，Workflow 启动时通过输入指定。