Go语言App Engine任务队列：如何创建和添加任务

package myapp

import (
    "fmt"
    "net/http"
    "time"

    "google.golang.org/appengine"
    "google.golang.org/appengine/taskqueue"
)

// handleAddTask 是一个HTTP处理函数，用于演示如何创建和添加任务
func handleAddTask(w http.ResponseWriter, r *http.Request) {
    c := appengine.NewContext(r)

    // 1. 创建 taskqueue.Task 对象
    // Path 字段是必不可少的，它指定了任务被执行时请求的目标URL
    task := &taskqueue.Task{
        Path: "/worker/process_data", // 指向处理该任务的HTTP处理函数
    }

    // 2. 将任务添加到默认队列
    // taskqueue.Add 返回一个 *Task 和一个 error
    addedTask, err := taskqueue.Add(c, task, "") // "" 表示使用默认队列
    if err != nil {
        http.Error(w, fmt.Sprintf("Failed to add task: %v", err), http.StatusInternalServerError)
        return
    }

    fmt.Fprintf(w, "Task '%s' added to queue. ETA: %s", addedTask.Name, addedTask.ETA.Format(time.RFC3339))
}

// handleWorkerProcessData 是一个HTTP处理函数，模拟任务的实际处理逻辑
// 这个处理函数应该在 /worker/process_data 路径上注册
func handleWorkerProcessData(w http.ResponseWriter, r *http.Request) {
    c := appengine.NewContext(r)
    // 实际的任务处理逻辑，例如从请求中读取数据，执行计算等
    // 任务请求通常是POST请求，任务的参数会放在请求体中
    fmt.Fprintf(w, "Worker task processed successfully!")
    c.Infof("Worker task /worker/process_data executed.")
}

func init() {
    http.HandleFunc("/add_task", handleAddTask)
    http.HandleFunc("/worker/process_data", handleWorkerProcessData)
}

package myapp

import (
    "fmt"
    "net/http"
    "time"

    "google.golang.org/appengine"
    "google.golang.org/appengine/datastore"
    "google.golang.org/appengine/taskqueue"
)

// handleTransactionAddTask 演示如何在数据存储事务中添加任务
func handleTransactionAddTask(w http.ResponseWriter, r *http.Request) {
    c := appengine.NewContext(r)

    err := datastore.RunInTransaction(c, func(txc appengine.Context) error {
        // 假设这里有一些数据存储操作
        // 例如：创建一个新的实体
        type MyEntity struct {
            Name string
            Time time.Time
        }
        entity := &MyEntity{
            Name: "Transactional Item",
            Time: time.Now(),
        }
        key := datastore.NewIncompleteKey(txc, "MyEntity", nil)
        _, err := datastore.Put(txc, key, entity)
        if err != nil {
            return fmt.Errorf("failed to put entity: %v", err)
        }

        // 在事务内部创建并添加任务
        // 重要的是，这里使用的是事务的上下文 `txc`
        task := &taskqueue.Task{
            Path: "/worker/process_transaction_data",
            // 可以在任务中传递参数，例如通过Payload或Header
            Payload: []byte("entity_created"),
            Method:  "POST", // 任务处理器通常期望POST请求
        }

        addedTask, err := taskqueue.Add(txc, task, "") // 使用事务上下文 txc
        if err != nil {
            return fmt.Errorf("failed to add task within transaction: %v", err)
        }

        txc.Infof("Task '%s' prepared for transaction commit.", addedTask.Name)
        return nil // 如果返回nil，事务将尝试提交
    }, nil) // 第二个nil表示使用默认事务选项

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

    fmt.Fprintf(w, "Transaction committed successfully, task will be executed.")
}

// handleWorkerProcessTransactionData 模拟处理事务任务的逻辑
func handleWorkerProcessTransactionData(w http.ResponseWriter, r *http.Request) {
    c := appengine.NewContext(r)
    // 从请求体中读取Payload
    payload := r.FormValue("Payload") // taskqueue.Add 默认将Payload作为表单值发送
    if payload == "" {
        // 如果Payload是原始字节，可能需要从请求体中直接读取
        // 或者将Payload设置为Header或URL参数
        payload = "No specific payload received."
    }
    fmt.Fprintf(w, "Worker task for transaction processed. Payload: %s", payload)
    c.Infof("Worker task /worker/process_transaction_data executed with payload: %s", payload)
}

func init() {
    http.HandleFunc("/add_transaction_task", handleTransactionAddTask)
    http.HandleFunc("/worker/process_transaction_data", handleWorkerProcessTransactionData)
}

taskWithDelay := &taskqueue.Task{
    Path:    "/worker/delayed_task",
    Payload: []byte("some_data_to_process"),
    Delay:   5 * time.Minute, // 任务将在5分钟后执行
    Method:  "POST",
}
_, err := taskqueue.Add(c, taskWithDelay, "my-custom-queue") // 添加到名为 "my-custom-queue" 的队列