Go语言在Google App Engine中创建任务队列任务的实践指南

import (
    "appengine/taskqueue"
)

// ... 在一个包含 appengine.Context 的函数中
func createTask(c appengine.Context) *taskqueue.Task {
    // 创建一个最基本的任务，指定其将被发送到的 worker URL 路径
    t := &taskqueue.Task{Path: "/path/to/your/worker/task"}
    return t
}

package myapp

import (
    "appengine"
    "appengine/datastore"
    "appengine/taskqueue"
    "fmt"
    "net/http"
)

// 假设有一个处理请求的HTTP handler
func mainHandler(w http.ResponseWriter, r *http.Request) {
    c := appengine.NewContext(r)
    if err := addTransactionalTask(c); err != nil {
        http.Error(w, fmt.Sprintf("Failed to add task: %v", err), http.StatusInternalServerError)
        return
    }
    fmt.Fprint(w, "Task initiated successfully.")
}

// addTransactionalTask 演示了如何在数据存储事务中添加任务
func addTransactionalTask(c appengine.Context) error {
    // 使用 datastore.RunInTransaction 运行一个事务
    err := datastore.RunInTransaction(c, func(txc appengine.Context) error {
        // 在此处执行数据存储操作，例如：
        // 假设有一个实体类型 MyEntity
        // type MyEntity struct {
        //     Name string
        //     Value int
        // }
        // entity := &MyEntity{Name: "TransactionData", Value: 123}
        // _, err := datastore.Put(txc, datastore.NewIncompleteKey(txc, "MyEntity", nil), entity)
        // if err != nil {
        //     c.Errorf("Datastore Put failed: %v", err)
        //     return err
        // }

        // 创建任务实例，指定 worker 处理器的路径
        t := &taskqueue.Task{Path: "/worker/process_data"}

        // 关键点：使用事务的上下文 (txc) 来添加任务。
        // 第三个参数是队列名称，"" 表示默认队列。
        _, err := taskqueue.Add(txc, t, "")
        if err != nil {
            c.Errorf("Failed to add task within transaction: %v", err)
            return err // 返回错误将导致事务回滚，任务也不会被添加
        }
        c.Infof("Task '%s' added successfully within transaction.", t.Path)
        return nil // 返回 nil 表示事务可以提交
    }, nil) // 最后一个参数是可选的事务选项

    if err != nil {
        c.Errorf("Transaction failed or rolled back: %v", err)
        return err
    }
    c.Infof("Transaction completed successfully.")
    return nil
}

import (
    "appengine"
    "appengine/taskqueue"
    "encoding/json"
    "time"
)

// TaskPayload 定义了任务的负载结构
type TaskPayload struct {
    UserID string `json:"user_id"`
    Action string `json:"action"`
}

func createComplexTask(c appengine.Context) (*taskqueue.Task, error) {
    payloadData := TaskPayload{UserID: "user456", Action: "generate_report"}
    payloadBytes, err := json.Marshal(payloadData)
    if err != nil {
        return nil, fmt.Errorf("failed to marshal payload: %w", err)
    }

    task := &taskqueue.Task{
        Path:    "/worker/process_report",
        Method:  "POST",
        Payload: payloadBytes,
        Header:  map[string][]string{"Content-Type": {"application/json"}},
        Delay:   10 * time.Minute, // 任务将在10分钟后执行
        Name:    fmt.Sprintf("report-task-%s-%d", payloadData.UserID, time.Now().Unix()), // 唯一任务名称
    }
    return task, nil
}

# app.yaml
runtime: go116 # 或您使用的Go版本

handlers:
# 匹配所有 /worker/ 开头的请求，并将其路由到Go应用
- url: /worker/.*
  script: auto # 对于Go 1.11+，使用 auto
  login: admin # 可选：限制只有管理员或App Engine系统可以访问此路径，增强安全性

package worker

import (
    "appengine"
    "fmt"
    "io/ioutil"
    "net/http"
    "encoding/json" // 如果任务负载是JSON
)

func init() {
    http.HandleFunc("/worker/process_data", processDataHandler)
    http.HandleFunc("/worker/process_report", processReportHandler)
}

// processDataHandler 处理 /worker/process_data 路径的任务
func processDataHandler(w http.ResponseWriter, r *http.Request) {
    c := appengine.NewContext(r)
    if r.Method != "POST" { // 默认任务方法是POST
        http.Error(w, "Method Not Allowed", http.StatusMethodNotAllowed)
        return
    }

    // App Engine会在请求头中添加任务相关信息，例如：
    // X-AppEngine-TaskName: 任务名称
    // X-AppEngine-RetryCount: 重试次数
    taskName := r.Header.Get("X-AppEngine-TaskName")
    c.Infof("Received task: %s, Retry Count: %s", taskName, r.Header.Get("X-AppEngine-RetryCount"))

    // 在此处执行具体的业务逻辑
    // ...

    fmt.Fprint(w, "Task processed successfully.") // 返回 200 OK 表示任务成功
}

// processReportHandler 处理 /worker/process_report 路径的任务，并解析JSON负载
func processReportHandler(w http.ResponseWriter, r *http.Request) {
    c := appengine.NewContext(r)
    if r.Method != "POST" {
        http.Error(w, "Method Not Allowed", http.StatusMethodNotAllowed)
        return
    }

    body, err := ioutil.ReadAll(r.Body)
    if err != nil {
        c.Errorf("Failed to read request body: %v", err)
        http.Error(w, "Bad Request", http.StatusBadRequest)
        return
    }

    var payload TaskPayload // 假设 TaskPayload 已在其他地方定义
    if err := json.Unmarshal(body, &payload); err != nil {
        c.Errorf("Failed to unmarshal task payload: %v", err)
        http.Error(w, "Bad Request", http.StatusBadRequest)
        return
    }

    c.Infof("Processing report for UserID: %s, Action: %s", payload.UserID, payload.Action)

    // 执行报告生成逻辑
    // ...

    fmt.Fprint(w, "Report task processed successfully.")
}