使用Go、App Engine和任务队列实现大规模高并发计数器

package frontend

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

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

// handleVote 接收用户投票请求，并将其作为任务添加到拉取队列
func handleVote(w http.ResponseWriter, r *http.Request) {
    ctx := appengine.NewContext(r)

    // 假设投票数据是一个简单的JSON字符串，包含用户ID、投票项等
    // 实际应用中，这里会解析请求体获取投票详情
    votePayload := []byte(fmt.Sprintf(`{"user_id": "%s", "item_id": "%s", "timestamp": %d}`,
        "user_"+r.RemoteAddr, "item_A", time.Now().UnixNano()))

    // 创建一个新任务，payload为投票数据，并可选择添加标签
    t := taskqueue.NewTask(votePayload, taskqueue.PullTag("vote_event"))
    // t.Header.Add("X-Custom-Header", "value") // 也可以添加自定义Header

    // 将任务添加到名为 "vote-pull-queue" 的拉取队列
    if _, err := taskqueue.Add(ctx, t, "vote-pull-queue"); err != nil {
        log.Printf("Failed to add vote task: %v", err)
        http.Error(w, "Failed to record vote", http.StatusInternalServerError)
        return
    }
    fmt.Fprintln(w, "Vote recorded successfully (task added to queue)")
}

func init() {
    http.HandleFunc("/vote", handleVote)
}

package worker

import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "net/http"
    "time"

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

// VotePayload 模拟投票任务的Payload结构
type VotePayload struct {
    UserID    string `json:"user_id"`
    ItemID    string `json:"item_id"`
    Timestamp int64  `json:"timestamp"`
}

// ShardedCounter 表示Datastore中的一个分片计数器实体
type ShardedCounter struct {
    Count int `datastore:"count"`
}

// leaseAndProcessVotesHandler 是工作者服务处理任务的HTTP处理函数
func leaseAndProcessVotesHandler(w http.ResponseWriter, r *http.Request) {
    ctx := appengine.NewContext(r)

    // 从 "vote-pull-queue" 队列中租赁最多100个任务，租赁期为10分钟
    tasks, err := taskqueue.Lease(ctx, 100, "vote-pull-queue", 10*time.Minute)
    if err != nil {
        log.Printf("Failed to lease tasks: %v", err)
        http.Error(w, "Failed to lease tasks", http.StatusInternalServerError)
        return
    }

    if len(tasks) == 0 {
        fmt.Fprintln(w, "No tasks to process.")
        return
    }

    log.Printf("Leased %d tasks for processing.", len(tasks))

    // 模拟处理任务并更新分片计数器
    // 实际应用中，这里会根据业务逻辑进行更复杂的聚合和分片更新

    // 示例：将所有投票计入一个随机选择的分片，或者根据时间/ID分片
    // 这里为了简化，我们假设只有一个分片 "shard_0"
    shardID := "shard_0" 
    counterKey := datastore.NewKey(ctx, "VoteCounter", shardID, 0, nil)

    // 在事务中更新计数器，确保原子性
    err = datastore.RunInTransaction(ctx, func(txCtx context.Context) error {
        var counter ShardedCounter
        // 尝试获取现有计数器，如果不存在则会创建
        err := datastore.Get(txCtx, counterKey, &counter)
        if err != nil && err != datastore.ErrNoSuchEntity {
            return fmt.Errorf("failed to get counter for shard %s: %w", shardID, err)
        }

        counter.Count += len(tasks) // 批量增加计数
        _, err = datastore.Put(txCtx, counterKey, &counter)
        if err != nil {
            return fmt.Errorf("failed to put counter for shard %s: %w", shardID, err)
        }
        return nil
    }, nil)

    if err != nil {
        log.Printf("Transaction failed for shard %s: %v", shardID, err)
        // 如果事务失败，任务将不会被删除，并在租赁期满后重新可见
        http.Error(w, "Failed to update counter due to transaction error", http.StatusInternalServerError)
        return
    }

    // 成功处理并更新计数后，删除任务
    if err := taskqueue.Delete(ctx, tasks); err != nil {
        log.Printf("Failed to delete processed tasks: %v", err)
        // 如果删除失败，任务会在租赁期满后重新可见，需要确保处理逻辑的幂等性
        http.Error(w, "Processed tasks but failed to delete them", http.StatusInternalServerError)
        return
    }

    fmt.Fprintf(w, "Successfully processed %d tasks and updated counter for shard %s.", len(tasks), shardID)
}

func init() {
    // 此处理函数通常由App Engine Cron Job定时触发，以确保任务被持续处理
    http.HandleFunc("/worker/process-votes", leaseAndProcessVotesHandler)
}