laravel 队列——迹忆客-ag捕鱼王app官网

频率限制

注：该功能要求应用可以与 redis 服务器进行交互。

如果应用使用了 redis，那么可以使用时间或并发来控制队列任务。该功能特性在队列任务与有频率限制的 api 交互时很有帮助。

例如，通过 throttle 方法，可以限定给定类型任务每 60 秒只运行 10 次。如果不能获取锁，需要将任务释放回队列以便可以再次执行：

redis::throttle('key')->allow(10)->every(60)->then(function () {
    // job logic...
}, function () {
    // could not obtain lock...
    return $this->release(10);
});

注：在上面的例子中，key 可以是任意可以唯一标识你想要限定访问频率的任务类型的字符串。举个例子，这个键可以基于任务类名和操作 eloquent 模型的 id 进行构建。

注：将受限制的任务释放回队列依然会增加任务的总执行次数 attempts 的值。

除此之外，还可以指定可以同时处理给定任务的最大进程数量。这个功能在队列任务正在编辑一次只能由一个任务进行处理的资源时很有用。例如，使用 funnel 方法你可以给定类型任务一次只能由一个工作进程进行处理：

redis::funnel('key')->limit(1)->then(function () {
    // job logic...
}, function () {
    // could not obtain lock...
    return $this->release(10);
});

注：使用频率限制时，任务在运行成功之前需要的最大尝试次数很难权衡，因此，将频率限制和基于时间的尝试次数结合起来使用是个不错的选择。

处理错误

如果任务在处理的时候有异常抛出，则该任务将会被自动释放回队列以便再次尝试执行。任务会持续被释放直到尝试次数达到应用允许的最大次数。最大尝试次数通过 artisan 命令 queue:work 上的 --tries 开关来定义。此外，该次数也可以在任务类自身上定义。关于运行队列监听器的更多信息可以在下面看到。

队列闭包

除了将任务类推送到队列之外，还可以推送闭包到队列。这对于需要在当前请求生命周期之外执行的简单快捷的任务来说非常方便：

$podcast = app\podcast::find(1);
dispatch(function () use ($podcast) {
    $podcast->publish();
});

推送闭包到队列时，闭包的代码内容以加密方式签名，所以不会在传输过程中被篡改。

运行队列进程

laravel 自带了一个队列进程用来处理被推送到队列的新任务。可以使用 queue:work 命令运行这个队列进程。请注意，队列进程开始运行后，会持续监听队列，直至手动停止或关闭终端：

$ php artisan queue:work

注：为了保持队列进程 queue:work 持续在后台运行，需要使用进程守护程序，比如 supervisor 来确保队列进程持续运行。

请记住，队列进程是长生命周期的进程，会在启动后驻留内存。若应用有任何改动将不会影响到已经启动的进程。所以请在发布程序后，重启队列进程。

此外，你还可以运行 queue:listen 命令。使用该命令时，代码修改后不需要手动重启队列进程，不过，该命令性能不及 queue:work：

$ php artisan queue:listen

指定连接和队列

队列进程同样可以自定义连接和队列。传递给 work 命令的连接名需要与配置文件 config/queue.php 中定义的某个连接配置相匹配：

$ php artisan queue:work redis

我们可以自定义将某个队列进程指定某个连接来管理。举例来说，如果所有的邮件任务都是通过 redis 连接上的 emails 队列处理，那么可以用以下命令来启动单一进程只处理单一队列：

$ php artisan queue:work redis --queue=emails

处理单个任务

--once 选项可用于告知进程只处理队列中的单个任务：

$ php artisan queue:work --once

处理所有队列任务然后退出

--stop-when-empty 选项可用于告知进程处理所有任务然后优雅退出。当我们在 docker 容器中处理 laravel 队列时，如果你想要在队列为空时关闭容器，则该选项很有用：

$ php artisan queue:work --stop-when-empty

资源注意事项

后台队列进程不会再处理每个任务前重启框架，因此需要在每次任务完成后释放所有重量级的资源。例如，如果我们在使用 gd 库处理图片，需要在完成的时候使用 imagedestroy 来释放内存。

队列优先级

有时候我们需要区分任务的优先级。比如，在配置文件 config/queue.php 中，可以定义 redis 连接的默认 queue 为 low。不过，如果需要将任务分发到高优先级 high，可以这么做：

dispatch((new job)->onqueue('high'));

如果期望所有 high 高优先级的队列都将先于 low 低优先级的任务执行，可以像这样启动队列进程：

$ php artisan queue:work --queue=high,low

队列进程 & 部署

前文已经提到队列进程是长生命周期的进程，在重启以前，所有源码的修改并不会对其产生影响。所以，最简单的方法是在每次发布新版本后重新启动队列进程。你可以通过 aritisan 命令 queue:restart 来优雅地重启队列进程:

$ php artisan queue:restart

该命令将在队列进程完成正在进行的任务后，结束该进程，避免队列任务的丢失或错误。由于队列进程会在执行 queue:restart 命令后死掉，你仍然需要通过进程守护程序如supervisor 来自动重启队列进程。

注：队列使用缓存来存储重启信号，所以在使用此功能前你需要验证缓存驱动配置正确。

任务过期 & 超时

任务过期

在配置文件 config/queue.php 中，每个连接都定义了 retry_after 项。该配置项的目的是定义任务在执行以后多少秒后释放回队列。如果retry_after 设定的值为 90, 任务在运行 90 秒后还未完成，那么将被释放回队列而不是删除掉。毫无疑问，你需要把 retry_after 的值设定为任务执行时间的最大可能值。

注：只有 amazon sqs 配置信息不包含 retry_after 项。amazon sqs 的任务执行时间基于 default visibility timeout ，该项在 amazon aws 控制台配置。

队列进程超时

队列进程 queue:work 可以设定超时 --timeout 项。该 --timeout 控制队列进程执行每个任务的最长时间，如果超时，该进程将被关闭。各种错误都可能导致某个任务处于“冻结”状态，比如 http 无响应等。队列进程超时就是为了将这些“冻结”的进程关闭：

$ php artisan queue:work --timeout=60

配置项 retry_after 和 aritisan 参数项 --timeout 不同，但目的都是为了确保任务的安全，并且只被成功的执行一次。

注：参数项 --timeout 的值应该始终小于配置项 retry_after 的值，这是为了确保队列进程总在任务重试以前关闭。如果 --timeout 比retry_after 大，那么你的任务可能被执行两次。

进程休眠时间

当任务在队列中有效时，进程会持续处理任务，没有延迟。不过，我们可以使用 sleep 配置项来指定没有新的有效任务产生时的休眠时间。休眠期间，队列进程不会处理任何新任务直到队列进程醒来：

$ php artisan queue:work --sleep=3

配置 supervisor

安装 supervisor

supervisor 是 linux 系统中常用的进程守护程序。如果队列进程 queue:work 意外关闭，它会自动重启启动队列进程。在 ubuntu 安装supervisor 非常简单：

$ sudo apt-get install supervisor

注：如果自己配置 supervisor 有困难，可以考虑使用 laravel forge，它会为 laravel 项目自动安装并配置 supervisor。

配置 supervisor

supervisor 配置文件通常存放在 /etc/supervisor/conf.d 目录，在该目录下，可以创建多个配置文件指示 supervisor 如何监视进程，例如，让我们创建一个开启并监视 queue:work 进程的 laravel-worker.conf 文件：

[program:laravel-worker]
process_name=%(program_name)s_%(process_num)02d
command=php /path/to/project/artisan queue:work sqs --sleep=3 --tries=3
autostart=true
autorestart=true
user=forge
numprocs=8
redirect_stderr=true
stdout_logfile=/home/forge/app.com/worker.log
stopwaitsecs=3600

注：我们需要将 /path/to/project 替换成自己项目在服务器的绝对地址。

在本例中，numprocs 指令让 supervisor 运行 8 个 queue:work 进程并监视它们，如果失败的话自动重启。当然，你需要修改 queue:work sqs 的 command 指令来映射你的队列连接。

注：我们需要确保 stopwaitsecs 的值大于最长运行任务的秒数，否则的话，supervisor 会在该任务处理完成前杀死对应的任务进程。

启动 supervisor

当成功创建配置文件后，需要刷新 supervisor 的配置信息并使用如下命令启动进程:

$ sudo supervisorctl reread
$ sudo supervisorctl update
$ sudo supervisorctl start laravel-worker:*

我们可以通过 获取更多信息。

处理失败的任务

不可避免会出现运行失败的任务。你不必为此担心，laravel 可以轻松设置任务允许的最大尝试次数，若是执行次数达到该限定，该任务会被插入到 failed_jobs 表，要创建一个 failed_jobs 表的迁移，可以使用 queue:failed-table 命令

$ php artisan queue:failed-table
$ php artisan migrate

然后，运行队列进程时，通过 --tries 参数项来设置队列任务允许的最大尝试次数，如果没有指定 --tries 选项的值，任务会被无限期重试：

$ php artisan queue:work redis --tries=3

此外，我们还可以通过 --delay 选项指定重试失败任务之前需要等待多少秒，默认情况下，失败队列任务会立即重试：

$ php artisan queue:work redis --tries=3 --delay=3

如果我们想要为指定任务配置失败重试延迟时间，可以在对应的队列任务类中定义 retryafter 属性：

/**
 * the number of seconds to wait before retrying the job.
 *
 * @var int
 */
public $retryafter = 3;

如果我们需要更复杂的逻辑来确定重试延迟，可以在排队的作业类上定义一个retryafter方法：

/**
* calculate the number of seconds to wait before retrying the job.
*
* @return int
*/
public function retryafter()
{
    return 3;
}

清理失败的任务

我们可以在任务类中定义 failed 方法, 从而允许在失败发生时执行指定的动作，比如发送任务失败的通知，记录日志等。导致任务失败的 exception 会被传递到failed 方法:

podcast = $podcast;
    }
    /**
     * execute the job.
     *
     * @param  audioprocessor  $processor
     * @return void
     */
    public function handle(audioprocessor $processor)
    {
        // process uploaded podcast...
    }
    /**
     * the job failed to process.
     *
     * @param  exception  $exception
     * @return void
     */
    public function failed(exception $exception)
    {
        // send user notification of failure, etc...
    }
}

注：如果任务通过 dispatchnow 方法分发的话，failed 方法不会调用。

任务失败事件

如果期望在任务失败的时候触发某个事件，可以使用 queue::failing 方法。该事件通过邮件或slack通知团队。举个例子，我么可以在 laravel 自带的 appserviceprovider 中添加一个回调到该事件：

connectionname
            // $event->job
            // $event->exception
        });
    }
}

重试失败的任务

要查看已插入到 failed_jobs 数据表中的所有失败任务，可以使用 artisan 命 queue:failed：

$ php artisan queue:failed

该命令将会列出任务 id、连接、队列和失败时间，任务 id 可用于重试失败任务，例如，要重试一个 id 为 5 的失败任务，可以运行下面的命令：

$ php artisan queue:retry 5

如有必要，我们可以将多个 id 或 id 范围（当使用数字 id 时）传递给命令：

$ php artisan queue:retry 5 6 7 8 9 10
$ php artisan queue:retry --range=5-10

要重试所有失败任务，运行如下命令即可:

$ php artisan queue:retry all

如果你要删除一个失败任务，可以使用 queue:forget 命令：

$ php artisan queue:forget 5

要删除所有失败任务，可以使用 queue:flush 命令：

$ php artisan queue:flush

忽略默认的模型

当注入一个 eloquent 模型到队列任务时，它会在被放到队列之前自动序列化，然后在任务被处理时恢复。不过，如果该模型实例在任务等待被处理期间被删除，对应任务在执行时会失败并抛出 modelnotfoundexception 异常。

为了方便起见，你可以通过设置队列任务的 deletewhenmissingmodels 属性为 true 来选择自动删除缺失模型实例的任务：

/**
 * delete the job if its models no longer exist.
 *
 * @var bool
 */
public $deletewhenmissingmodels = true;

任务事件

通过 queue facade 提供的 before 和 after 方法可以在任务被处理之前或之后指定要执行的回调。这些回调可用来记录日志或者记录统计数据。通常，我们可以在服务提供者中使用这些方法。比如，我们可以在appserviceprovider 中这样用:

connectionname
            // $event->job
            // $event->job->payload()
        });
        queue::after(function (jobprocessed $event) {
            // $event->connectionname
            // $event->job
            // $event->job->payload()
        });
    }
}

使用 queue facade上的 looping 方法，可以在进程尝试从队列中获取任务之前指定要执行的回调。例如，可以注册一个闭包来回滚之前失败任务遗留下来的事务：

queue::looping(function () {
    while (db::transactionlevel() > 0) {
        db::rollback();
    }
});