Go App Engine高并发分片计数器实践：利用任务队列构建可靠投票系统

package main

import (
    "context"
    "log"
    "time"

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

// submitVote 模拟用户提交投票，将投票项ID作为任务推送到拉取队列
func submitVote(ctx context.Context, itemID string) error {
    // 任务的Payload可以是一个简单的字符串，也可以是JSON编码的复杂结构
    payload := []byte(itemID) 

    // 创建一个拉取任务
    t := &taskqueue.Task{
        Payload: payload,
        Method:  "PULL", // 明确指定为拉取任务
    }

    // 将任务添加到名为 "my-pull-queue" 的队列中
    _, err := taskqueue.Add(ctx, t, "my-pull-queue")
    if err != nil {
        log.Printf("ERROR: Failed to add vote task for item %s: %v", itemID, err)
        return err
    }
    log.Printf("INFO: Vote task for item %s added to queue.", itemID)
    return nil
}

// 示例用法
func main() {
    ctx := appengine.NewContext(nil) // 获取App Engine上下文
    err := submitVote(ctx, "item_A")
    if err != nil {
        // 处理错误
    }
    err = submitVote(ctx, "item_B")
    if err != nil {
        // 处理错误
    }
    // ... 更多投票
}

package main

import (
    "context"
    "log"
    "math/rand"
    "strconv"
    "time"

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

const (
    numShards = 10 // 每个投票项的分片数量
    queueName = "my-pull-queue"
)

// CounterShard 定义Datastore中计数器分片的结构
type CounterShard struct {
    Count int `datastore:"count"`
}

// processVotesWorker 模拟Worker服务周期性处理投票任务
func processVotesWorker(ctx context.Context) {
    // 租用最多100个任务，租期为1小时
    // 租期内，其他Worker不能租用这些任务
    tasks, err := taskqueue.LeaseTasks(ctx, 100, queueName, 1*time.Hour)
    if err != nil {
        log.Printf("ERROR: Failed to lease tasks: %v", err)
        return
    }

    if len(tasks) == 0 {
        log.Printf("INFO: No tasks to process.")
        return
    }

    log.Printf("INFO: Leased %d tasks.", len(tasks))

    // 用于存储每个投票项的聚合计数
    itemVoteCounts := make(map[string]int)

    // 遍历租用的任务，聚合计数
    for _, t := range tasks {
        itemID := string(t.Payload) // 假设Payload是投票项ID
        itemVoteCounts[itemID]++
    }

    // 更新Datastore中的分片计数器
    err = updateShardedCounters(ctx, itemVoteCounts)
    if err != nil {
        log.Printf("ERROR: Failed to update sharded counters: %v", err)
        // 注意：如果更新失败，这些任务不会被删除，租期结束后会重新变为可用，
        // 从而实现自动重试。Worker应具备幂等性。
        return
    }

    // 成功更新Datastore后，删除已处理的任务
    err = taskqueue.DeleteTasks(ctx, queueName, tasks...)
    if err != nil {
        log.Printf("ERROR: Failed to delete tasks: %v", err)
        // 即使删除失败，任务在租期结束后也会重新可用，Worker的幂等性很重要
    } else {
        log.Printf("INFO: Successfully processed and deleted %d tasks.", len(tasks))
    }
}

// updateShardedCounters 负责更新Datastore中的分片计数器
func updateShardedCounters(ctx context.Context, counts map[string]int) error {
    for itemID, increment := range counts {
        // 随机选择一个分片进行更新，以分散写入负载
        shardID := rand.Intn(numShards)
        shardKey := datastore.NewKey(ctx, "CounterShard", itemID+"_shard_"+strconv.Itoa(shardID), 0, nil)

        // 使用事务来保证计数器更新的原子性
        err := datastore.RunInTransaction(ctx, func(tx *datastore.Transaction) error {
            var shard CounterShard
            err := tx.Get(shardKey, &shard)
            if err != nil && err != datastore.ErrNoSuchEntity {
                return err
            }
            shard.Count += increment
            _, err = tx.Put(shardKey, &shard)
            return err
        }, nil) // 默认重试选项

        if err != nil {
            log.Printf("ERROR: Failed to update shard for item %s, shard %d: %v", itemID, shardID, err)
            return err // 返回错误，让上层决定是否重试整个批次
        }
    }
    return nil
}

// 示例用法：通常由App Engine Cron Job或另一个Worker服务触发
func main() {
    ctx := appengine.NewContext(nil)
    // 这是一个简化的循环，实际应用中Worker会作为一个长期运行的服务，
    // 可能通过定时触发或持续循环来拉取任务。
    for {
        processVotesWorker(ctx)
        time.Sleep(5 * time.Second) // 间隔一段时间再次尝试拉取任务
    }
}