有没有办法确保 GAE 上的任务队列具有 FIFO（先进先出）行为？答案

【问题标题】：Is there a way to assure FIFO (first in, first out) behavior with Task Queues on GAE?有没有办法确保 GAE 上的任务队列具有 FIFO（先进先出）行为？
【发布时间】：2012-04-16 11:08:10
【问题描述】：

有没有办法通过 GAE 上的任务队列确保 FIFO（先进先出）行为？

GAE 文档说 FIFO 是影响任务执行顺序的因素之一，但同一份文档说“系统的调度可能会‘跳’新任务到队列的头部”，我已经确认了这种行为测试。效果：我的事件被乱序处理了。

文档说：

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

任务执行的顺序取决于几个因素：

任务在队列中的位置。 App Engine 尝试根据 FIFO >（先进先出）顺序处理任务。通常，任务被插入到队列的末尾，并且从队列的头部执行。

队列中的任务积压。系统尝试提供最低延迟通过对调度程序的特别优化的通知，任何给定的任务都是可能的。因此，如果队列有大量任务积压， 系统的调度可能会将新任务“跳转”到队列的头部。

任务的 etaMillis 属性的值。此属性指定任务可以执行的最早时间。 App Engine 总是等到在指定的 ETA 之后处理推送任务。

任务的 countdownMillis 属性的值。此属性指定最小值执行任务前等待的秒数。倒计时和 eta 是互斥的；如果您指定一个，请不要指定另一个。

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

我现在有什么？

由消费者调用并创建任务的 Rest servlet（事件数据在有效负载上）。
在这之后，一个worker servlet得到这个Task并且：
- 反序列化事件数据；
- 将事件放在数据存储上；
- 在数据存储上更新车辆。

那么，再一次，有什么方法可以保证 FIFO 的行为？或者我该如何改进这个解决方案来获得这个？

【问题讨论】：

为什么需要严格的 FIFO？请记住，在分布式系统中，事件的顺序有点模糊 - 当您的前端和后端分布在多台机器上时，即使告诉几个几乎同时发生的请求中的哪一个最先发生也是困难的（而且通常毫无意义）。作为最终结果，您要达到什么目标？
我们跟踪公共巴士，所以我们需要知道它什么时候停在一个公共汽车站，什么时候开始旅行，或者什么时候超过了速度限制。问题是一些事件与之前的事件相关，因为我们也做状态管理。例如：如果公共汽车在之前的事件中记录了“关闭的行程”，则只能“开启行程”。所以，你可以想象如果我把这些事件弄乱了会发生什么......
任务队列在任何情况下听起来都不是最好的方法。您要做的是使用数据存储事务，并将给定总线的状态转换集限制为状态机允许的状态转换。
在没有任务队列的维护或应用更新的情况下，如何停止接收事件？
我不明白你在问什么。您通常如何接收活动更新？

标签： java google-app-engine task-queue

【解决方案1】：

您需要通过三个单独的步骤来解决此问题：

实现 Sharding Counter 以单调生成增加ID。尽管我喜欢使用来自的timestamp 谷歌的服务器来指示任务排序，似乎时间戳 GAE 服务器之间的差异可能会超出您的要求。
将您的任务添加到 Pull Queue 而不是 Push Queue。什么时候构建您的TaskOption，将步骤#1 中获得的ID 添加为tag。添加任务后，将ID 存储在数据存储中的某个位置。
让您的工作 servlet lease Tasks by a certain tag 来自 Pull Queue。查询数据存储以获取您需要获取的最早 ID，并使用 ID 作为租约tag。通过这种方式，您可以为您的任务队列模拟 FIFO 行为。

完成处理后，请从您的数据存储中删除 ID，同时不要忘记从您的 Pull Queue 中删除 Task。另外，我建议您在后端运行您的任务消耗。

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

无论生成ID的方式如何，基本思想是使用datastore的查询机制产生Tasks的FIFO排序，并使用任务Tag从TaskQueue拉取特定任务。

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

【讨论】：

真的可以使用分片计数器创建单调递增的 id 吗？我认为不是 - 在低级 api 数据存储中分配 id 的函数可以生成 id，但我怀疑可能不是严格的顺序。
@mjaggard：谢谢，但我不太确定我理解你的意思。分片计数器不使用低级数据存储中的生成 ID 功能。基本上，我们自己增加计数器，但我们将“计数器桶”分布在多个实体上以减少写入争用。因此，计数器将单调递增。当然，请 cmiiw。
是的，您不能使用分片计数器生成单调递增的 ID - 分片计数器跨越多个实体组，因此无法保证您生成的任何数字都是唯一的。分片计数器用于计算事物，而不是用于分配 ID。
@NickJohnson：感谢 Nick 的更正，这让我感到惊讶，因为我在生产应用程序上使用分片计数器代替时间戳来对实体进行时间排序。我可能会就此提出一个新问题来澄清这种行为。
@IbrahimArief 分片计数器从未用于分配 ID。为什么不直接使用时间戳？

【解决方案2】：

我认为简单的答案是“不”，但部分是为了帮助改善这种情况，我正在使用拉队列 - 一次拉 1000 个任务，然后对它们进行排序。如果时间不重要，您可以对它们进行排序并将它们放入数据存储区，然后一次完成一批。您仍然需要弄清楚如何处理批处理开始和结束时的任务 - 因为它们可能会因其他批处理中的交错任务而出现故障。

【讨论】：

时机很重要。忘了告诉我需要按时间戳顺序处理，这是在车辆内的嵌入式设备上生成的。

【解决方案3】：

好的。我就是这样做的。

1) Rest servlet that is called from the consumer:

    If Event sequence doesn't match Vehicle sequence (from datastore)

        Creates a task on a "wait" queue to call me again

    else

       State validation

       Creates a task on the "regular" queue (Event data is on payload).


2) A worker servlet gets the task from the "regular" queue, and so on... (same pseudo code)

这样我可以暂停“常规”队列，以便在不丢失事件的情况下进行数据维护。

感谢您的回答。我的解决方案是混合使用它们。

【讨论】：

【解决方案4】：

您可以使用创建时间戳将要完成的工作放在数据存储中的一行中，然后按该时间戳获取工作任务，但如果您的任务创建得太快，您将遇到延迟问题。

【讨论】：

【解决方案5】：

我自己不知道答案，但使用延迟函数排队的任务可能会按提交的顺序执行。您可能需要 G. 的工程师才能得到答案。按照建议拉取队列似乎是一个不错的选择，而且这将允许您考虑批量处理 put()。

关于分片计数器的一点说明：它们会增加 id 单调增加的概率，但不能保证它们。

【讨论】：

deferred 使用任务队列。
对，只是不知道从延迟函数与标准入队进入队列的任务是否有一些魔力。听起来答案是“不”，这是我猜到的。

【解决方案6】：

处理此问题的最佳方式（分布式方式或“App Engine 方式”）可能是修改您的算法和数据收集以仅使用时间戳，从而允许对任务进行任意排序。

假设这不可能或太难，你可以修改你的算法如下：

在创建任务时，不要将数据放在有效负载上，而是放在数据存储区中，以按时间戳排序并存储为您尝试更新的任何实体的子实体（车辆？） .时间戳应该来自客户端，而不是服务器，以保证相同的顺序。
运行一个通用任务，获取第一个时间戳的数据，处理它，然后在事务中删除它。

【讨论】：

【解决方案7】：

在此线程之后，我不清楚严格的 FIFO 要求是针对所有收到的交易，还是针对每辆车。后者比前者有更多选择。

【讨论】：