Google AppEngine：任务队列API

任务队列

com.google.appengine.api.taskqueue

使用任务队列，用户可以发起一个请求，以使应用程序执行此请求之外的工作。它们是进行后台工作的强大工具。

此外，您可以将工作组织成小的离散单元（任务）。然后，应用程序根据队列的配置将这些任务插入一个或多个队列，并以FIFO顺序进行处理。这是我从Google IO演示中获得的图表，该图表从较高级别说明了将任务插入队列中的情况：

队列配置

1.推送队列（默认）：

推送队列将根据队列定义中配置的处理速率来处理任务（请参见下文）。 App Engine自动管理这些队列的生存期（创建，删除等），并调整处理能力以匹配您的配置和处理量。这些只能在App Engine（您的应用内部）中使用。

2.拉入队列：

允许任务使用者在特定时间范围内的特定时间租用任务。可通过Task Queue REST API在内部和外部对其进行访问。但是，在这种情况下，GAE不会自动管理队列的生命周期和处理速率，这要由开发人员来决定。后端也可以访问这些队列。

任务

它们代表应用程序执行的工作单元。问是幂等的，即它们在队列中是唯一的，根据Google文档，不能同时调用多次（除非发生一些奇怪的内部错误情况）。

TaskOptions类的实例，任务由URL和有效载荷组成，有效载荷可以是简单的字符串，二进制对象（byte []）或DeferredTask的实例。 DeferredTask本质上是可运行的。这使您可以将任务链接在一起。我们的团队必须这样做，以便在GAE的最大执行限制为30秒时模拟长时间运行的任务。当前，任务必须完成执行并在原始请求后的10分钟内发送200-299之间的HTTP响应值。该期限与用户请求分开，后者的期限为60秒。

此外，要求使用令牌桶来控制任务执行的速率。每次调用任务时，都会使用一个令牌。这种租赁模型（获取令牌）通常是代理系统或消息传递系统的模型，它允许用户控制这些任务的执行速率（请参阅下文中的配置队列）。

最后，Task Queue API的一个非常重要的功能是它具有自动重试任务的功能。创建TaskOptions对象时，可以使用RetriesOptions参数进行配置。

交易中的任务

任务可以作为数据存储事务的一部分入队。如果事务成功提交，将确保插入（不执行）。唯一需要注意的是，事务性任务不能具有用户定义的名称，并且在单个事务中最多有5个插入到任务队列中。

组态

队列是通过queue.xml配置的。如果省略，则使用具有默认配置的默认队列。由于请求队列满足更高级的需求，因此必须对其进行专门配置（没有默认的请求队列）。

应用程序的队列配置适用于该应用程序的所有版本。您可以使用queue.xml中的target参数为推送队列覆盖此行为。如果您想要具有不同队列处理配置的不同版本的应用程序（不同的站点），则使用此方法。

这是允许您配置的一些内容（文档更详尽）：

• bucket-size ：当队列中有许多任务且速率很高时（仅推送），处理队列的速度。（警告：开发服务器将忽略此值）

• max-concurrent-requests ：在指定队列中的任何给定时间可以执行的最大任务数（仅推送）。

• 模式：是推还是拉。

• 名称：队列名称

• rate ：在此队列上处理任务的频率（s =秒，m =分钟，h =小时，d =天）。如果为0，则认为队列已暂停。（警告：开发服务器将忽略此值）

• 目标：将任务定位到特定的后端或应用程序版本。

<queue-entries> 
<!--Set the number of max concurrent requests to 10-->   <queue>     <name>optimize-queue</name>                 <rate>20/s</rate>   <bucket-size>40</bucket-size>       <max-concurrent-requests>10</max-concurrent-requests>     </queue> 
</queue-entries>

样例代码

这是一个非常简单的例子。如前所述，任务队列基本上是一个URL处理程序。在此Servlet中，GET将处理入队任务。该任务将POST到相同的servlet，并执行执行任务的doPost（）方法。在这种情况下，它只是一个简单的计数器。请注意，计数器是一个易失性属性。如果您将此Servlet作为GET请求访问，它将排队另一个任务。因此，您将看到两个任务都将计数器增加。

public class TaskQInfo extends HttpServlet {private static volatile int TASK_COUNTER = 0;// Executed by user menu clickpublic void doGet(HttpServletRequest req, HttpServletResponse resp)throws IOException {// Build a task using the TaskOptions Builder pattern from ** aboveQueue queue = QueueFactory.getDefaultQueue();queue.add(withUrl("/taskq_demo").method(TaskOptions.Method.POST)); resp.getWriter().println("Task have been added to default queue...");resp.getWriter().println("Refresh this page to add another count task");}// Executed by TaskQueue@Overrideprotected void doPost(HttpServletRequest req, HttpServletResponse resp)throws ServletException, IOException {// This is the body of the taskfor(int i = 0; i < 1000; i++) {log.info("Processing: " + req.getHeader("X-AppEngine-TaskName") + "-" +           TASK_COUNTER++); try { // Sleep for a second (if the rate is set to 1/s this will allow at // most 1 more task to be processed)Thread.sleep(1000); } catch (InterruptedException e) { // ignore}}}
}

任务队列使您可以通过按需调用后台进程来在应用程序中实现某种程度的并发。对于非常冗长的任务，您可能需要查看App Engine 后端，这些后端基本上是没有请求时间限制的特殊App Engine实例。

参考： Google AppEngine：来自JCG合作伙伴 Luis Atencio的Task Queues API ，位于Reflective Thought博客上。