Google App Engine Go应用中高并发投票计数与数据一致性维护实践

package myapp

import (
    "context"
    "time"

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

// VoteEvent 代表一次投票事件
type VoteEvent struct {
    VoterID   string    `datastore:"voterId"`
    Option    string    `datastore:"option"`
    Timestamp time.Time `datastore:"timestamp"`
}

// saveVoteEvent 将投票事件保存到Datastore
func saveVoteEvent(ctx context.Context, voterID, option string) error {
    vote := &VoteEvent{
        VoterID:   voterID,
        Option:    option,
        Timestamp: time.Now(),
    }
    // 创建一个不带Key的实体，Datastore会自动生成ID
    _, err := datastore.Put(ctx, datastore.NewIncompleteKey(ctx, "VoteEvent", nil), vote)
    if err != nil {
        log.Errorf(ctx, "Failed to save vote event: %v", err)
        return err
    }
    log.Infof(ctx, "Vote event saved: %s for %s", voterID, option)
    return nil
}

package myapp

import (
    "context"
    "fmt"

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

// enqueueRecalculationTask 将重算任务添加到队列
func enqueueRecalculationTask(ctx context.Context) error {
    task := taskqueue.NewTask("/tasks/recalculate-votes", nil)
    // 为任务指定一个固定名称，确保同一时间只有一个重算任务
    task.Name = "recalculate-all-votes"

    _, err := taskqueue.Add(ctx, task, "") // "" 表示默认队列
    if err != nil {
        // 如果任务已存在，则忽略错误
        if err == taskqueue.ErrTaskAlreadyExists {
            log.Infof(ctx, "Recalculation task already enqueued.")
            return nil
        }
        log.Errorf(ctx, "Failed to enqueue recalculation task: %v", err)
        return err
    }
    log.Infof(ctx, "Recalculation task enqueued.")
    return nil
}

// 投票处理函数示例
func handleVote(ctx context.Context, voterID, option string) error {
    err := saveVoteEvent(ctx, voterID, option)
    if err != nil {
        return fmt.Errorf("error saving vote: %v", err)
    }

    // 投票成功后，触发重算任务
    err = enqueueRecalculationTask(ctx)
    if err != nil {
        // 即使任务添加失败，原始投票数据也已保存，可以考虑重试机制或报警
        log.Warningf(ctx, "Could not enqueue recalculation task: %v", err)
    }
    return nil
}

package myapp

import (
    "context"
    "net/http"

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

// VoteCounts 存储每个选项的投票总数
type VoteCounts struct {
    Option string `datastore:"option"`
    Count  int    `datastore:"count"`
}

// RecalculateVotesHandler 是任务队列的HTTP处理程序
func RecalculateVotesHandler(w http.ResponseWriter, r *http.Request) {
    ctx := r.Context()

    // 1. 查询所有原始投票事件
    var voteEvents []VoteEvent
    keys, err := datastore.NewQuery("VoteEvent").GetAll(ctx, &voteEvents)
    if err != nil {
        log.Errorf(ctx, "Failed to fetch vote events: %v", err)
        http.Error(w, "Internal Server Error", http.StatusInternalServerError)
        return
    }

    // 2. 统计票数
    counts := make(map[string]int)
    for _, event := range voteEvents {
        counts[event.Option]++
    }

    // 3. 更新或创建投票总数实体
    var entitiesToSave []*VoteCounts
    var keysToSave []*datastore.Key

    for option, count := range counts {
        // 尝试获取现有计数实体
        countKey := datastore.NewKey(ctx, "VoteCounts", option, 0, nil)
        currentCount := &VoteCounts{}
        err := datastore.Get(ctx, countKey, currentCount)

        if err == datastore.ErrNoSuchEntity {
            // 如果不存在，则创建新的
            currentCount = &VoteCounts{Option: option, Count: count}
            entitiesToSave = append(entitiesToSave, currentCount)
            keysToSave = append(keysToSave, countKey)
        } else if err != nil {
            log.Errorf(ctx, "Failed to get existing vote count for %s: %v", option, err)
            http.Error(w, "Internal Server Error", http.StatusInternalServerError)
            return
        } else {
            // 如果存在，则更新
            currentCount.Count = count
            entitiesToSave = append(entitiesToSave, currentCount)
            keysToSave = append(keysToSave, countKey)
        }
    }

    // 批量保存更新后的计数
    if len(entitiesToSave) > 0 {
        _, err = datastore.PutMulti(ctx, keysToSave, entitiesToSave)
        if err != nil {
            log.Errorf(ctx, "Failed to save vote counts: %v", err)
            http.Error(w, "Internal Server Error", http.StatusInternalServerError)
            return
        }
    }

    log.Infof(ctx, "Vote counts recalculated successfully.")
    w.WriteHeader(http.StatusOK)
}

// 在 app.yaml 中配置路由
// handlers:
// - url: /tasks/recalculate-votes
//   script: _go_app
//   login: admin # 确保只有App Engine内部可以调用此URL