README

本文描述了异步任务队列的工作原理。

安装

您可以将任务队列作为标准TAO扩展添加到当前TAO实例。

 $ composer require oat-sa/extension-tao-task-queue

队列组件

队列可以与不同类型的队列代理一起工作，以下两种类型用于实现异步机制

• RdsQueueBroker，该代理将任务存储在RDS中。
• SqsQueueBroker，用于使用AWS SQS。

注意:

当使用SqsQueueBroker时，请确保"oat-sa/lib-generis-aws"包含在主composer.json中，并且您已经正确配置了generis/awsClient.conf.php。

权重

队列可以具有权重。如果正在使用多个队列，则此权重将用于随机选择要消费的队列。例如，如果队列A的权重为1，队列B的权重为2，则队列B有大约66%的概率被选中。

工作组件

这里有一个所谓的LongRunningWorker，它可以无限期地运行。它内置了对以下操作的信号处理

• 优雅地关闭工作进程：SIGTERM/SIGINT/SIGQUIT
• 暂停任务处理：SIGUSR2
• 恢复任务处理：SIGCONT

注意:

可以同时运行多个工作进程。

处理完指定任务后，工作进程通过任务日志保存任务生成的报告。

服务设置示例

多个队列设置

在这种情况下，我们注册了3个队列：其中一个是使用SQS代理，另外两个是RDS。每个队列都有自己的权重（如90、30、10），这将用于在选择下一个要消费的队列时使用。

我们有两个任务与不同的队列相关联，此外，默认队列已指定（'background'），它将用于在OPTION_TASK_TO_QUEUE_ASSOCIATIONS中未定义的所有其他任务。

use oat\tao\model\taskQueue\QueueDispatcher;
use oat\tao\model\taskQueue\Queue;
use oat\taoTaskQueue\model\QueueBroker\RdsQueueBroker;
use oat\taoTaskQueue\model\QueueBroker\SqsQueueBroker;
use oat\tao\model\taskQueue\TaskLogInterface;
use oat\tao\model\taskQueue\QueueDispatcherInterface;

$queueService = new QueueDispatcher(array(
    QueueDispatcherInterface::OPTION_QUEUES => [
        new Queue('priority', new SqsQueueBroker('default', \common_cache_Cache::SERVICE_ID, 10), 90),
        new Queue('standard', new RdsQueueBroker('default', 5), 30),
        new Queue('background', new RdsQueueBroker('default', 5), 10)
    ],
    QueueDispatcherInterface::OPTION_TASK_LOG     => TaskLogInterface::SERVICE_ID,
    QueueDispatcherInterface::OPTION_TASK_TO_QUEUE_ASSOCIATIONS => [
        SomeImportantAction::class => 'priority',
        SomeLessImportantTask::class => 'standard'
    ]
));

$queueService->setOption(QueueDispatcherInterface::OPTION_DEFAULT_QUEUE, 'background');

$this->getServiceManager()->register(QueueDispatcherInterface::SERVICE_ID, $queueService);

如果队列尚未初始化，即尚未创建所需的队列容器

try {
    $queueService->initialize();
} catch (\Exception $e) {
    return \oat\oatbox\reporting\Report::createError('Initializing queues failed');
}

初始化队列和任务日志容器

如果您想确保创建所需的队列和任务日志容器，则可以运行此脚本。

 $ sudo -u www-data php index.php 'oat\taoTaskQueue\scripts\tools\InitializeQueue'

注意:

此脚本也可以用于将当前队列更改为使用不同的队列代理。

• 将每个现有队列更改为使用InMemoryQueueBroker。 (同步队列)

 $ sudo -u www-data php index.php 'oat\taoTaskQueue\scripts\tools\InitializeQueue' --broker=memory

• 将每个现有队列更改为使用RdsQueueBroker。需要"持久性"选项，"接收"（在轮询队列时可以接收的最大任务数）是可选的。

 $ sudo -u www-data php index.php 'oat\taoTaskQueue\scripts\tools\InitializeQueue' --broker=rds --persistence=default --receive=10

• 将每个现有队列更改为使用SqsQueueBroker。需要"aws-profile"选项，"接收"是可选的。

 $ sudo -u www-data php index.php 'oat\taoTaskQueue\scripts\tools\InitializeQueue' --broker=sqs --aws-profile=default --receive=10

• 如果您想将上述设置应用于特定队列，请在命令中添加--queue=...选项。在以下情况下，仅修改myQueue。

 $ sudo -u www-data php index.php 'oat\taoTaskQueue\scripts\tools\InitializeQueue' --queue=myQueue --broker=rds --persistence=default --receive=10

• 设置任务选择策略。

 $ sudo -u www-data php index.php 'oat\taoTaskQueue\scripts\tools\InitializeQueue' --strategy="\oat\taoTaskQueue\model\TaskSelector\StrictPriorityStrategy"

运行工作进程

要运行工作进程，请使用以下命令。它将启动一个无限期运行的工作进程，并基于其权重遍历每个注册的队列。

 $ sudo -u www-data php index.php 'oat\taoTaskQueue\scripts\tools\RunWorker'

如果您想为专用队列运行工作进程，请将队列名称传递给命令，如下所示

 $ sudo -u www-data php index.php 'oat\taoTaskQueue\scripts\tools\RunWorker' --queue=priority

您可以限制工作进程的迭代次数。这只适用于专用队列。

 $ sudo -u www-data php index.php 'oat\taoTaskQueue\scripts\tools\RunWorker' --queue=standard --limit=5

如果您想将特定任务关联到新队列，可以使用此命令

 $ sudo -u www-data php index.php 'oat\taoTaskQueue\scripts\tools\ManageAssociationMap' \
      -t  '{ you fully qualified task class name }' -q queue-name

下次定义的任务创建时，它将被分配到指定的队列。

总结停滞任务

如果您想总结停滞任务，请执行此命令。例如

sudo -u www-data php index.php 'oat\taoTaskQueue\scripts\tools\StuckTaskSummary' \
--queue indexation_queue \
--age 300 \
--whitelist "oat\tao\model\search\tasks\UpdateResourceInIndex,oat\tao\model\search\tasks\UpdateClassInIndex"

重启停滞任务

如果您想重启停滞任务，请执行此命令。例如

sudo -u www-data php index.php 'oat\taoTaskQueue\scripts\tools\StuckTaskRestart' \
--queue indexation_queue \
--age 300 \
--whitelist "oat\tao\model\search\tasks\UpdateResourceInIndex,oat\tao\model\search\tasks\UpdateClassInIndex"

REST API

任务日志报告可以通过应用程序编程接口（API）查看/消费。为了使用它，请检查（doc/taskApi.yml）中的swagger文件。

命令行工具

除了使用API检查任务报告之外，还可以使用命令行。

sudo -u www-data php index.php 'oat\taoTaskQueue\scripts\tools\TaskLogUtility' --help

此命令将显示工具可以执行的所有可能操作。

Examples
 1. Stats
	 Description: 	 Return stats about the tasks logs statuses
	 Example: 	 sudo -u www-data php index.php 'oat\taoTaskQueue\scripts\tools\TaskLogUtility' --stats
 2. List Task Logs
	 Description: 	 List All the tasks that are not archived will be retrived, default limit is 20
	 Example: 	 sudo -u www-data php index.php 'oat\taoTaskQueue\scripts\tools\TaskLogUtility' --available --limit[optional]=20 --offset[optional]=10
 3. Get Task Log
	 Description: 	 Get an specific task log by id
	 Example: 	 sudo -u www-data php index.php 'oat\taoTaskQueue\scripts\tools\TaskLogUtility' --get-task=[taskdId]
 4. Archive a Task Log
	 Description: 	 Archive a task log
	 Example: 	 sudo -u www-data php index.php 'oat\taoTaskQueue\scripts\tools\TaskLogUtility' --archive=[taskdId] --force[optional]
 5. Cancel a Task Log
	 Description: 	 Cancel a task log
	 Example: 	 sudo -u www-data php index.php 'oat\taoTaskQueue\scripts\tools\TaskLogUtility' --cancel=[taskdId] --force[optional]