Go语言App Engine任务队列：任务创建与事务集成

import (
    "appengine/taskqueue"
)

// 创建一个最基本的任务实例
// Path 字段指定了处理这个任务的HTTP端点
t := &taskqueue.Task{
    Path: "/path/to/workertask", // 你的任务处理程序URL
}

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

func enqueueHandler(w http.ResponseWriter, r *http.Request) {
    c := appengine.NewContext(r)

    // 创建任务
    t := &taskqueue.Task{Path: "/path/to/workertask"}

    // 将任务添加到默认队列
    _, err := taskqueue.Add(c, t, "") // "" 表示使用默认队列
    if err != nil {
        c.Errorf("Failed to add task: %v", err)
        http.Error(w, fmt.Sprintf("Error adding task: %v", err), http.StatusInternalServerError)
        return
    }

    fmt.Fprintln(w, "Task successfully enqueued!")
}

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

func transactionAndTaskHandler(w http.ResponseWriter, r *http.Request) {
    c := appengine.NewContext(r)

    err := datastore.RunInTransaction(c, func(txc appengine.Context) error {
        // --- 1. 在事务中执行数据存储操作 ---
        // 假设这里有一些数据存储操作，例如保存一个实体
        key := datastore.NewIncompleteKey(txc, "MyEntity", nil)
        _, err := datastore.Put(txc, key, &struct{ Value string }{Value: "Transactional Data"})
        if err != nil {
            return fmt.Errorf("failed to put entity in transaction: %v", err)
        }
        txc.Infof("Entity put successfully within transaction.")

        // --- 2. 在事务中创建并添加任务 ---
        // 创建任务，指定其处理路径
        t := &taskqueue.Task{
            Path: "/path/to/workertask",
            // 可以在这里添加任务参数，例如通过Form字段
            // Form: url.Values{"entity_key": {key.Encode()}},
        }

        // ！！！关键：使用事务提供的上下文 (txc) 来添加任务 ！！！
        _, err = taskqueue.Add(txc, t, "") // "" 表示使用默认队列
        if err != nil {
            return fmt.Errorf("failed to add task within transaction: %v", err)
        }
        txc.Infof("Task added successfully within transaction.")

        return nil // 事务成功
    }, nil) // 第二个nil可以用于指定事务选项，这里使用默认值

    if err != nil {
        c.Errorf("Transaction failed: %v", err)
        http.Error(w, fmt.Sprintf("Transaction failed: %v", err), http.StatusInternalServerError)
        return
    }

    fmt.Fprintln(w, "Operation successful: data saved and task enqueued atomically.")
}

package myapp

import (
    "appengine"
    "fmt"
    "net/http"
    "time" // 示例中使用
)

func init() {
    // 注册任务处理程序
    http.HandleFunc("/path/to/workertask", workerTaskHandler)
}

// workerTaskHandler 是处理任务队列任务的HTTP处理程序
func workerTaskHandler(w http.ResponseWriter, r *http.Request) {
    c := appengine.NewContext(r)

    // 建议验证请求是否来自任务队列，增强安全性
    // 可以检查请求头 "X-AppEngine-TaskName" 或 "X-AppEngine-QueueName"
    if r.Header.Get("X-AppEngine-TaskName") == "" {
        c.Warningf("Received non-taskqueue request to worker endpoint.")
        http.Error(w, "Forbidden", http.StatusForbidden)
        return
    }

    taskName := r.Header.Get("X-AppEngine-TaskName")
    queueName := r.Header.Get("X-AppEngine-QueueName")
    c.Infof("Received task '%s' from queue '%s'.", taskName, queueName)

    // 从请求中获取任务数据（如果任务中包含Payload或Form数据）
    // 例如，如果任务是通过Form字段传递数据：
    // myData := r.FormValue("data_key")
    // c.Infof("Task data: %s", myData)

    // --- 执行实际的后台工作 ---
    c.Infof("Starting heavy lifting for task %s...", taskName)
    time.Sleep(5 * time.Second) // 模拟耗时操作
    c.Infof("Finished heavy lifting for task %s.", taskName)

    // 任务处理成功，返回200 OK
    // 如果返回非2xx状态码，App Engine会认为任务失败并可能重试
    w.WriteHeader(http.StatusOK)
    fmt.Fprintln(w, "Task processed successfully.")
}

import (
    "appengine/taskqueue"
    "net/url"
    "time"
)

// 创建一个带参数并延迟执行的任务
taskWithParams := &taskqueue.Task{
    Path:  "/path/to/workertask",
    Form:  url.Values{"user_id": {"12345"}, "action": {"process_order"}},
    Delay: 10 * time.Minute, // 任务将在10分钟后执行
    Name:  "process-order-12345-" + time.Now().Format("20060102150405"), // 唯一任务名
}

// 然后使用 taskqueue.Add(c, taskWithParams, "my-custom-queue") 添加到队列