使用Go、App Engine和任务队列实现高吞吐量分片计数器

package main

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

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

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

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

    // 假设投票内容是简单的用户ID或投票项ID
    votePayload := []byte(fmt.Sprintf("user_id:%s, item_id:%s", r.FormValue("userId"), r.FormValue("itemId")))

    // 创建一个新任务
    t := taskqueue.NewTask(votePayload, 0) // payload是投票数据，0表示默认延迟

    // 将任务添加到名为 "vote-pull-queue" 的拉取队列
    // 确保在app.yaml或queue.yaml中定义了此队列为拉取队列
    _, err := taskqueue.Add(ctx, t, "vote-pull-queue")
    if err != nil {
        log.Printf("Failed to add task to queue: %v", err)
        http.Error(w, "Failed to record vote temporarily", http.StatusInternalServerError)
        return
    }

    w.WriteHeader(http.StatusAccepted)
    fmt.Fprintln(w, "Vote received and queued for processing.")
}

package main

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

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

// 定义计数器实体结构
type Shard struct {
    Count int `datastore:"count"`
}

func init() {
    http.HandleFunc("/process-votes", processVotesHandler)
}

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

    // 从拉取队列租用任务
    // LeaseTasks参数：队列名称，最大任务数，租用时长
    tasks, err := taskqueue.LeaseTasks(ctx, "vote-pull-queue", 1000, 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的哈希值进行分片
    // 假设我们有10个Memcache分片，键为 "vote_count_shard_0" 到 "vote_count_shard_9"
    shardCounts := make(map[int]int) // 存储每个分片的增量

    for _, t := range tasks {
        // 解析任务payload，提取投票信息
        // 例如：votePayload := string(t.Payload)
        // 实际应用中可能需要更复杂的解析，例如JSON或Protobuf
        _ = t.Payload // 假设我们只是简单计数，不关心具体内容
        shardKey := time.Now().Second() % 10 // 简单示例：按秒的哈希值分片，实际应更稳定
        shardCounts[shardKey]++
    }

    // 批量更新Memcache分片
    for shardID, increment := range shardCounts {
        memcacheKey := fmt.Sprintf("vote_count_shard_%d", shardID)
        _, err := memcache.IncrementExisting(ctx, memcacheKey, int64(increment))
        if err != nil && err != memcache.ErrCacheMiss { // 如果键不存在，则初始化
            item := &memcache.Item{
                Key:        memcacheKey,
                Value:      []byte(fmt.Sprintf("%d", increment)),
                Expiration: 24 * time.Hour, // 根据需求设置过期时间
            }
            err = memcache.Add(ctx, item)
            if err != nil {
                log.Printf("Failed to add initial memcache item %s: %v", memcacheKey, err)
                // 错误处理：可以考虑将这些任务重新放回队列或记录下来
            }
        } else if err == memcache.ErrCacheMiss {
            // 如果是第一次增量，需要先设置值
            item := &memcache.Item{
                Key:        memcacheKey,
                Value:      []byte(fmt.Sprintf("%d", increment)),
                Expiration: 24 * time.Hour,
            }
            err = memcache.Add(ctx, item)
            if err != nil {
                log.Printf("Failed to add initial memcache item %s: %v", memcacheKey, err)
            }
        }
    }

    // 批量删除已处理的任务
    if err := taskqueue.DeleteTasks(ctx, "vote-pull-queue", tasks...); err != nil {
        log.Printf("Failed to delete tasks: %v", err)
        // 严重错误：任务未删除，可能导致重复处理。需要有机制处理这种情况，例如幂等性设计。
        http.Error(w, "Failed to delete tasks after processing", http.StatusInternalServerError)
        return
    }

    fmt.Fprintln(w, "Votes processed and counters updated.")
}

// 示例：从Memcache读取并更新Datastore的Cron处理函数
func persistCountersHandler(w http.ResponseWriter, r *http.Request) {
    ctx := appengine.NewContext(r)

    // 遍历所有Memcache分片键
    for i := 0; i < 10; i++ { // 假设有10个分片
        memcacheKey := fmt.Sprintf("vote_count_shard_%d", i)
        item, err := memcache.Get(ctx, memcacheKey)
        if err != nil {
            if err == memcache.ErrCacheMiss {
                continue // 该分片无数据
            }
            log.Printf("Failed to get memcache item %s: %v", memcacheKey, err)
            continue
        }

        currentCount := 0
        fmt.Sscanf(string(item.Value), "%d", &currentCount)

        // 更新Datastore中的分片计数器
        shardKey := datastore.NewKey(ctx, "VoteShard", fmt.Sprintf("shard_%d", i), 0, nil)
        err = datastore.RunInTransaction(ctx, func(txCtx context.Context) error {
            var shard Shard
            if err := datastore.Get(txCtx, shardKey, &shard); err != nil && err != datastore.ErrNoSuchEntity {
                return err
            }
            shard.Count += currentCount
            _, err := datastore.Put(txCtx, shardKey, &shard)
            return err
        }, nil)

        if err != nil {
            log.Printf("Failed to update Datastore for shard %d: %v", i, err)
        } else {
            // 成功更新后，可以考虑将Memcache中的该分片计数清零或减去已持久化的值
            // 为了简单起见，这里选择不清零，而是让下一个周期继续增量，但需要注意重复计数问题
            // 更好的方法是使用memcache.CompareAndSwap或在事务中处理Memcache更新
            log.Printf("Shard %d updated in Datastore with %d votes.", i, currentCount)
        }
    }
    fmt.Fprintln(w, "Counters persisted to Datastore.")
}

# app.yaml
runtime: go118 # 或更高版本

instance_class: F2 # 适当的实例类型

handlers:
- url: /vote
  script: auto
  login: required # 示例：如果需要认证

- url: /process-votes
  script: auto
  target: worker-service # 假设处理任务的服务名为 worker-service
  login: admin # 仅限管理员访问，或通过内部调用

- url: /persist-counters
  script: auto
  target: cron-service # 假设持久化服务名为 cron-service
  login: admin # 仅限管理员访问，或通过内部调用

# 定义其他服务，例如 worker-service 和 cron-service
# worker-service/app.yaml
# runtime: go118
# instance_class: F2
# handlers:
# - url: /.*
#   script: auto

# cron-service/app.yaml
# runtime: go118
# instance_class: F1
# handlers:
# - url: /.*
#   script: auto

# queue.yaml
queue:
- name: vote-pull-queue
  mode: pull
  rate: 5/s # 示例：每秒允许5个任务被添加到队列，可以根据需求调整
  bucket_size: 100 # 示例：任务桶大小
  max_concurrent_leases: 100 # 示例：最大并发租用任务数

# cron.yaml
cron:
- description: "Persist vote counts to Datastore"
  url: /persist-counters
  target: cron-service
  schedule: every 5 minutes