Go语言App Engine任务队列任务创建指南

import (
    "appengine/taskqueue"
)

// ...

// 创建一个最简单的任务
task := &taskqueue.Task{
    Path: "/path/to/your/worker", // 任务处理器的URL路径
}

package myapp

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

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

// MyEntity 定义一个简单的数据结构用于演示
type MyEntity struct {
    Name      string
    Timestamp time.Time
}

func init() {
    http.HandleFunc("/", handleMain)
    http.HandleFunc("/worker/process-data", handleWorker) // 任务处理函数
}

// handleMain 处理主请求，创建数据并添加任务
func handleMain(w http.ResponseWriter, r *http.Request) {
    c := appengine.NewContext(r)

    // 在事务中创建实体并添加任务
    err := datastore.RunInTransaction(c, func(tc appengine.Context) error {
        // 1. 创建一个数据实体
        entity := &MyEntity{
            Name:      "Transaction Data " + time.Now().Format("15:04:05"),
            Timestamp: time.Now(),
        }
        key := datastore.NewIncompleteKey(tc, "MyEntity", nil)
        _, err := datastore.Put(tc, key, entity)
        if err != nil {
            return fmt.Errorf("failed to put entity: %v", err)
        }

        // 2. 创建并添加任务
        // 这是用户提问的核心部分：如何创建 taskqueue.Task 实例
        task := &taskqueue.Task{
            Path:    "/worker/process-data", // 指定处理任务的HTTP路径
            Payload: []byte(fmt.Sprintf("entity_name=%s", entity.Name)), // 可以携带数据
            Method:  "POST", // 任务默认使用 POST 方法
            // 其他可选字段：
            // Delay:   time.Second * 5, // 延迟执行
            // Header:  http.Header{"X-Custom-Header": {"Value"}},
        }

        // 使用事务的上下文 tc 来添加任务
        // 第二个参数 "" 表示将任务添加到默认队列
        _, err = taskqueue.Add(tc, task, "")
        if err != nil {
            return fmt.Errorf("failed to add task: %v", err)
        }

        tc.Infof("Successfully created entity and added task: %s", entity.Name)
        return nil
    }, nil) // nil 表示默认事务选项

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

    fmt.Fprintf(w, "Transaction successful! Check logs for task creation.")
}

// handleWorker 是任务队列的处理函数
// 这个函数会在 /worker/process-data 路径上接收并处理任务请求
func handleWorker(w http.ResponseWriter, r *http.Request) {
    c := appengine.NewContext(r)
    if r.Method != "POST" {
        http.Error(w, "Method Not Allowed", http.StatusMethodNotAllowed)
        return
    }

    // 从请求中获取任务数据
    // 如果 Payload 是 form-urlencoded 格式，可以使用 r.FormValue
    // 如果是 JSON 等其他格式，需要从 r.Body 读取并解析
    entityName := r.FormValue("entity_name")
    if entityName == "" {
        c.Errorf("Worker received empty entity_name")
        http.Error(w, "Bad Request: missing entity_name", http.StatusBadRequest)
        return
    }

    c.Infof("Worker received task for entity: %s. Processing...", entityName)
    // 在这里执行实际的业务逻辑，例如更新数据库、调用外部服务等
    // 模拟耗时操作
    time.Sleep(time.Second * 2)
    c.Infof("Worker finished processing task for entity: %s", entityName)

    w.WriteHeader(http.StatusOK)
    fmt.Fprintf(w, "Task processed successfully.")
}