java - 有没有办法确保 GAE 上任务队列的 FIFO(先进先出)行为？

Question

有没有办法确保 GAE 上任务队列的 FIFO(先进先出)行为？

GAE 文档说 FIFO 是影响任务执行顺序的因素之一，但同一份文档说“系统的调度可能会‘跳’新任务到队列的头部”，我已经通过测试。结果:我的事件正在乱序处理。

Docs says:

https://developers.google.com/appengine/docs/java/taskqueue/overview-push

The order in which tasks are executed depends on several factors:

The position of the task in the queue. App Engine attempts to process tasks based on FIFO > (first in, first out) order. In general, tasks are inserted into the end of a queue, and executed from the head of the queue.

The backlog of tasks in the queue. The system attempts to deliver the lowest latency possible for any given task via specially optimized notifications to the scheduler. Thus, in the case that a queue has a large backlog of tasks, the system's scheduling may "jump" new tasks to the head of the queue.

The value of the task's etaMillis property. This property specifies the earliest time that a task can execute. App Engine always waits until after the specified ETA to process push tasks.

The value of the task's countdownMillis property. This property specifies the minimum number of seconds to wait before executing a task. Countdown and eta are mutually exclusive; if you specify one, do not specify the other.

我需要做什么？在我的用例中，我每天将处理来自车辆的 1-2 百万个事件。这些事件可以以任何时间间隔(1 秒、1 分钟或 1 小时)发送。必须确保事件处理的顺序。 我需要按时间戳顺序进行处理，该顺序在车内的嵌入式设备上生成。

我现在有什么？

1. 由消费者调用并创建任务的 Rest servlet(事件数据在有效负载上)。

2. 在此之后，一个工作 servlet 得到这个任务并且:

   • 反序列化事件数据；

   • 将事件放在数据存储上；

   • 在数据存储上更新车辆。

那么，有没有什么方法可以确保 FIFO 行为呢？或者我怎样才能改进这个解决方案来获得这个？

Answer 1

您需要通过三个独立的步骤来解决这个问题:

1. 实现 Sharding Counter单调生成 增加 ID。尽管我喜欢使用 timestamp 来自 谷歌的服务器来指示任务排序，看来是时间戳 GAE 服务器之间的差异可能超出您的要求。

2. 将您的任务添加到 Pull Queue而不是 Push Queue .什么时候 构建您的 TaskOption，添加从步骤 #1 获得的 ID 作为 tag . 添加任务后，将 ID 存储在数据存储区的某个位置。

3. 让你的 worker servlet lease Tasks by a certain tag来自 Pull Queue。 查询数据存储以获取您需要获取的最早 ID，并将 ID 用作 租约标签。通过这种方式，您可以为您的任务队列模拟 FIFO 行为。

完成处理后，从数据存储中删除 ID，不要忘记从 Pull Queue 中删除 Task也。此外，我建议您在后端运行任务消费。

更新: 正如 Nick Johnson 和 mjaggard 所指出的，步骤 #1 中的分片似乎无法生成单调递增的 ID，因此需要其他 ID 来源。我似乎记得您使用的是车辆生成的时间戳，是否可以使用它来代替单调递增的 ID？

无论采用何种方式生成ID，其基本思路都是利用datastore的查询机制，生成Tasks的FIFO排序，并使用task的Tag拉取具体的来自 TaskQueue 的任务。

不过有一个警告。由于高复制数据存储的最终一致性读取策略，如果您选择 HRD 作为您的数据存储(并且您应该从 2012 年 4 月 4 日起弃用 M/S)，查询可能会返回一些陈旧数据第 2 步。