google-app-engine - 使用 go 遍历 appengine 上的大量实体

Question

在 App Engine 上，我有大量特定类型的实体。我想在每个实体上运行一个函数(例如编辑实体或复制它)我会在任务队列中执行此操作，但任务队列的运行时间限制为 10 分钟，并且每个函数调用都容易出现多种错误。执行此操作的最佳方法是什么？

Answer 1

这是我的解决方案，但我希望有人能提供更好的解决方案。我也想知道这是否容易产生 fork 炸弹，例如如果任务运行两次，它将引发两条迭代链..!我只用它来迭代几十万个实体，尽管对每个实体的操作都很昂贵。

首先，我创建了一个任务队列，用于一次运行一个实体上的每个函数调用:

queue:
- name: entity-iter
  rate: 100/s
  max_concurrent_requests: 1
  retry_parameters:
    task_retry_limit: 3
    task_age_limit: 30m
    min_backoff_seconds: 200

然后我有一个迭代实体方法，在给定类型的情况下，它将使用 key 在每个实体上调用您的延迟函数。

package sysadmin

import (
    "google.golang.org/appengine/datastore"
    "golang.org/x/net/context"
    "google.golang.org/appengine/log"
    "google.golang.org/appengine/delay"
    "google.golang.org/appengine/taskqueue"
)


func ForEachEntity(kind string, f *delay.Function) *delay.Function {
    var callWithNextKey *delay.Function // func(c context.Context, depth int, cursorString string) error
    callWithNextKey = delay.Func("something", func(c context.Context, depth int, cursorString string) error {
        q := datastore.NewQuery(kind).KeysOnly()
        if cursorString != "" {
            if curs, err := datastore.DecodeCursor(cursorString); err != nil {
                log.Errorf(c, "error decoding cursor %v", err)
                return err
            } else {
                q = q.Start(curs)
            }
        }
        it := q.Run(c)
        if key, err := it.Next(nil); err != nil {
            if err == datastore.Done {
                log.Infof(c, "Done %v", err)
                return nil
            }
            log.Errorf(c, "datastore error %v", err)
            return err
        } else {
            curs, _ := it.Cursor()
            if t, err := f.Task(key); err != nil {
                return err
            } else if _, err = taskqueue.Add(c, t, "entity-iter"); err != nil {
                log.Errorf(c, "error %v", err)
                return err
            }
            if depth - 1 > 0 {
                if err := callWithNextKey.Call(c, depth - 1, curs.String()); err != nil {
                    log.Errorf(c, "error2 %v", err)
                    return err
                }
            }
        }
        return nil
    })
    return callWithNextKey
}

示例用法:

var DoCopyCourse = delay.Func("something2", CopyCourse)

var DoCopyCourses = ForEachEntity("Course", DoCopyCourse)

func CopyCourses(c context.Context) {
    //sharedmodels.MakeMockCourses(c)
    DoCopyCourses.Call(c, 9999999, "")
}