README

此 Laravel/Lumen 包提供 Amazon SQS FIFO 队列的队列驱动程序。虽然 Laravel 默认支持 Amazon SQS 标准队列，但 FIFO 队列有所不同，且 Laravel 无法正确处理。这正是本包发挥作用的地方。

版本

本包已在 Laravel 4.1 至 Laravel 10.x 上进行了测试，尽管它可能在后续版本中继续工作。本节将更新以反映已测试的版本。

本说明已更新以显示最新支持版本（9.x - 10.x）的信息。对于 Laravel 4.1 至 Laravel 8.x，请查看 2.x 分支。

安装

通过 Composer

$ composer require shiftonelabs/laravel-sqs-fifo-queue

一旦 composer 已更新且包已安装，需要加载服务提供者。

Laravel 9.x, 10.x

此包使用自动包发现。服务提供者将自动注册。

Lumen 9.x, 10.x

打开 bootstrap/app.php 并在“注册服务提供者”部分下添加以下行

$app->register(ShiftOneLabs\LaravelSqsFifoQueue\LaravelSqsFifoQueueServiceProvider::class);

配置

Laravel/Lumen 9.x, 10.x

如果使用 Lumen，在项目根目录中创建一个 config 目录（如果您还没有的话）。然后，将 vendor/laravel/lumen-framework/config/queue.php 复制到 config/queue.php。

现在，对于 Laravel 和 Lumen，打开 config/queue.php 并将以下条目添加到 connections 数组中。

'sqs-fifo' => [
    'driver' => 'sqs-fifo',
    'key' => env('AWS_ACCESS_KEY_ID'),
    'secret' => env('AWS_SECRET_ACCESS_KEY'),
    'prefix' => env('SQS_FIFO_PREFIX', 'https://sqs.us-east-1.amazonaws.com/your-account-id'),
    'queue' => env('SQS_FIFO_QUEUE', 'default.fifo'),
    'suffix' => env('SQS_FIFO_SUFFIX'),
    'region' => env('AWS_DEFAULT_REGION', 'us-east-1'),
    'after_commit' => false,
    'group' => 'default',
    'deduplicator' => env('SQS_FIFO_DEDUPLICATOR', 'unique'),
    'allow_delay' => env('SQS_FIFO_ALLOW_DELAY', false),
],

示例 .env 文件

AWS_ACCESS_KEY_ID=ABCDEFGHIJKLMNOPQRST
AWS_SECRET_ACCESS_KEY=1a23bc/deFgHijKl4mNOp5qrS6TUVwXyz7ABCDef
SQS_FIFO_PREFIX=https://sqs.us-east-1.amazonaws.com/123456789012
SQS_FIFO_QUEUE=queuename.fifo
SQS_FIFO_SUFFIX=-staging
AWS_DEFAULT_REGION=us-east-1
SQS_FIFO_DEDUPLICATOR=unique
SQS_FIFO_ALLOW_DELAY=false

如果您希望这是默认连接，请在 .env 文件中将 QUEUE_CONNECTION=sqs-fifo 也设置为。

Capsule

如果在使用 Lumen/Laravel 之外使用 illuminate\queue 组件 Capsule

use Illuminate\Queue\Capsule\Manager as Queue;
use ShiftOneLabs\LaravelSqsFifoQueue\LaravelSqsFifoQueueServiceProvider;

$queue = new Queue;

$queue->addConnection([
    'driver' => 'sqs-fifo',
    'key'    => 'your-public-key',   // ex: ABCDEFGHIJKLMNOPQRST
    'secret' => 'your-secret-key',   // ex: 1a23bc/deFgHijKl4mNOp5qrS6TUVwXyz7ABCDef
    'prefix' => 'your-prefix',       // ex: https://sqs.us-east-1.amazonaws.com/your-account-id
    'queue' => 'your-queue-name',    // ex: queuename.fifo
    'suffix' => 'your-suffix',       // ex: -staging
    'region' => 'your-queue-region', // ex: us-east-1
    'after_commit' => false,
    'group' => 'default',
    'deduplicator' => 'unique',
    'allow_delay' => false,
], 'sqs-fifo');

// Make this Capsule instance available globally via static methods... (optional)
$queue->setAsGlobal();

// Register the 'queue' alias in the Container, then register the SQS FIFO provider.
$app = $queue->getContainer();
$app->instance('queue', $queue->getQueueManager());
(new LaravelSqsFifoQueueServiceProvider($app))->register();

凭证

如果使用 AWS 凭证提供的替代选项之一（例如，使用 AWS 凭证文件）来提供 AWS 凭证，则可以省略 key 和 secret 配置选项。有关替代选项的更多信息，请参阅AWS PHP SDK 指南。

AWS STS 会话令牌

如果需要指定 AWS STS 临时会话令牌，可以在配置中添加 'token' => env('AWS_SESSION_TOKEN'), 配置选项。这对于某些特定环境（例如 AWS Lambda）是必需的。

队列后缀

使用 suffix 配置选项可以在不指定环境后缀的情况下支持不同环境的队列。例如，如果您有一个 emails-staging.fifo 队列和一个 emails-production.fifo 队列，您可以根据环境将 suffix 配置设置为 -staging 或 -production，然后您的代码可以继续使用 emails.fifo 而无需知道环境。因此，Job::dispatch()->onQueue('emails.fifo') 将使用配置中定义的 suffix 所对应的 emails-staging.fifo 或 emails-production.fifo 队列。

注意：SQS FIFO 队列必须以 .fifo 后缀结尾。如上例所示，配置中定义的任何 suffix 都将出现在所需的 .fifo 后缀之前。不要在后缀配置或队列名称中指定 .fifo，否则队列名称将无法正确生成。

作业与数据库事务

使用 after_commit 配置选项来控制数据库事务中间排队的工作处理方式。当设置为 true 时，此选项确保在数据库事务提交之后才实际分发在数据库事务中排队的作业。如果事务回滚，作业将被丢弃而不会分发。如果没有活动数据库事务，作业将立即分发。

排队事件监听器

目前，此包无法处理排队事件监听器。为此需要接管 Laravel 的事件分发器，但这不是此包目前准备要做的事情。为了解决这个问题，你需要将排队监听器转换为非排队的监听器，然后分发一个新的排队作业来处理工作。

用法

大多数情况下，此队列驱动程序的用法与内置队列驱动程序相同。然而，在与 Amazon 的 SQS FIFO 队列一起工作时，有一些额外的事情需要考虑。

消息组

除了可以为每个连接拥有多个队列名称之外，SQS FIFO 队列还允许每个 FIFO 队列拥有多个“消息组”。这些消息组用于将相关的作业分组在一起，并且每个组内的作业按 FIFO 顺序处理。这很重要，因为你的队列性能可能取决于能否正确分配消息组。如果你有 100 个作业在队列中，并且它们都属于一个消息组，那么一次只能有一个队列工作者能够处理这些作业。然而，如果它们可以逻辑上分为 5 个消息组，那么你可以有 5 个队列工作者处理来自队列的作业（每个组一个）。FIFO 排序是按消息组进行的。

目前，默认情况下，所有排队作业将汇总到一个组中，如配置文件中定义。在上面的配置中，所有排队作业都将作为 default 组的一部分发送。可以使用 onMessageGroup() 方法更改每个作业的组，这将在下面进行解释。

组 ID 不能为空，不能超过 128 个字符，并且可以包含字母数字字符和标点符号（!"#$%&'()*+,-./:;<=>?@[\]^_`{|}~）。

在未来版本中，将能够将消息组分配给一个函数，如下面的去重器。

去重

向 SQS FIFO 队列发送作业时，Amazon 需要一种方法来确定作业是否是队列中已存在的作业的副本。SQS FIFO 队列有一个 5 分钟的去重间隔，这意味着如果在间隔内发送重复的消息，它将成功接受（无错误），但不会交付或处理。

确定重复消息通常有两种方式：要么所有消息都被视为独特，无论内容如何，要么如果消息具有相同的内容，则被视为重复。

此包通过 deduplicator 配置选项处理去重。

为了使所有消息都被视为独特，将 deduplicator 设置为 unique。

要考虑内容相同的消息为重复项，有两种选项，具体取决于FIFO队列的配置方式。如果FIFO队列在Amazon中已启用基于内容去重功能，则应将deduplicator设置为sqs。这表示连接将依赖于Amazon来确定内容唯一性。然而，如果禁用了基于内容去重功能，则应将deduplicator设置为content。注意，如果禁用了基于内容去重，而将deduplicator设置为sqs，则在尝试将作业发送到队列时将生成错误。

总结

sqs - 当需要将内容相同的消息视为重复项且在SQS FIFO队列上启用了基于内容去重时使用。
content - 当需要将内容相同的消息视为重复项，但在SQS FIFO队列上禁用了基于内容去重时使用。
unique - 当所有消息都应该被视为唯一，而不管内容如何时使用。

如果需要不同的去重算法，可以在容器中注册自定义去重方法。

最后，默认情况下，所有队列作业将使用配置文件中定义的去重器。可以使用withDeduplicator()方法按作业更改此设置。

延迟作业

SQS FIFO队列不支持按消息延迟，只支持按队列延迟。所需的延迟在Amazon控制台设置队列时定义在队列本身上。尝试为发送到FIFO队列的作业设置延迟将不会有任何效果。为了延迟作业，可以将作业push()到一个已定义了交付延迟的SQS FIFO队列。

由于不支持按消息延迟，使用later()方法将作业推送到SQS FIFO队列将默认抛出BadMethodCallException异常。但是，可以使用allow_delay配置选项来更改此行为。

将队列的allow_delay配置选项设置为true将允许later()方法将作业推送到该队列而不会抛出异常。但是，发送到later()方法的延迟参数将被完全忽略，因为延迟时间是在队列本身的SQS上定义的。

高级用法

按作业组和去重器

如果您需要更改特定作业的组或去重器，则需要访问onMessageGroup()和withDeduplicator()方法。这些方法通过ShiftOneLabs\LaravelSqsFifoQueue\Bus\SqsFifoQueueable特性提供。一旦将此特性添加到您的作业类中，就可以更改该特定作业的组或去重器，而不会影响队列上的任何其他作业。

代码示例

作业

<?php

namespace App\Jobs;

use Illuminate\Bus\Queueable;
use Illuminate\Queue\SerializesModels;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Contracts\Queue\ShouldQueue;
use ShiftOneLabs\LaravelSqsFifoQueue\Bus\SqsFifoQueueable;

class ProcessCoin implements ShouldQueue
{
    use InteractsWithQueue, Queueable, SqsFifoQueueable, SerializesModels;

    //
}

用法

dispatch(
    (new \App\Jobs\ProcessCoin())
        ->onMessageGroup('quarter')
        ->withDeduplicator('unique')
);

通知

<?php

namespace App\Notifications;

use Illuminate\Bus\Queueable;
use Illuminate\Notifications\Notification;
use Illuminate\Contracts\Queue\ShouldQueue;
use ShiftOneLabs\LaravelSqsFifoQueue\Bus\SqsFifoQueueable;

class InvoicePaid extends Notification implements ShouldQueue
{
    use Queueable, SqsFifoQueueable;

    //
}

用法

$user->notify(
    (new InvoicePaid($invoice))->onMessageGroup($invoice->id)
);

可邮寄

<?php

namespace App\Mail;

use Illuminate\Bus\Queueable;
use Illuminate\Mail\Mailable;
use Illuminate\Queue\SerializesModels;
use Illuminate\Contracts\Queue\ShouldQueue;
use ShiftOneLabs\LaravelSqsFifoQueue\Bus\SqsFifoQueueable;

class OrderShipped extends Mailable implements ShouldQueue
{
    use Queueable, SerializesModels, SqsFifoQueueable;

    //
}

用法

Mail::to($request->user())
    ->cc($moreUsers)
    ->bcc($evenMoreUsers)
    ->queue(new OrderShipped($order)->onMessageGroup($order->number));

自定义去重器

去重器通过生成发送到队列的去重ID来工作。如果两个消息生成了相同去重ID，则第二个消息被视为重复项，如果它在5分钟去重间隔内，则该消息将不会投递。

如果您有一些需要使用的自定义逻辑来生成去重ID，则可以注册自己的自定义去重器。去重器存储在IoC容器中，前缀为queue.sqs-fifo.deduplicator。例如，unique去重器被别名为queue.sqs-fifo.deduplicator.unique。

自定义去重器通过在IoC中注册一个新的带前缀的别名来创建。此别名应解析到实现ShiftOneLabs\LaravelSqsFifoQueue\Contracts\Queue\Deduplicator接口的新对象实例。您可以选择定义一个新的实现此接口的类，或者创建一个新的ShiftOneLabs\LaravelSqsFifoQueue\Queue\Deduplicators\Callback实例，该实例接受一个执行去重逻辑的Closure。定义的Closure应接受两个参数：$payload和$queue，其中$payload是要发送到队列的json_encoded()消息，而$queue是消息要发送到的队列名称。生成的ID不得超过128个字符，可以包含字母数字字符和标点符号（!"#$%&'()*+,-./:;<=>?@[\]^_`{|}~）。

例如，如果您想创建一个random去重器，该去重器会随机选择一些作业作为重复项，您可以在AppServiceProvider的register()方法中添加以下行

$this->app->bind('queue.sqs-fifo.deduplicator.random', function ($app) {
    return new \ShiftOneLabs\LaravelSqsFifoQueue\Queue\Deduplicators\Callback(function ($payload, $queue) {
        // Return the deduplication id generated for messages. Randomly 0 or 1.
        return mt_rand(0,1);
    });
}

或者，如果您更喜欢创建一个新的类，您的类可能看起来像这样

namespace App\Deduplicators;

use ShiftOneLabs\LaravelSqsFifoQueue\Contracts\Queue\Deduplicator;

class Random implements Deduplicator
{
    public function generate($payload, $queue)
    {
        // Return the deduplication id generated for messages. Randomly 0 or 1.
        return mt_rand(0,1);
    }
}

然后您可以在您的AppServiceProvider中这样注册该类

$this->app->bind('queue.sqs-fifo.deduplicator.random', App\Deduplicators\Random::class);

注册此别名后，您可以更新配置中的deduplicator键，使用值random，或者您可以通过在作业上调用withDeduplicator('random')来为单个作业设置去重器。

贡献

欢迎贡献。请参阅CONTRIBUTING以获取详细信息。

安全

如果您发现任何与安全相关的问题，请通过电子邮件patrick@shiftonelabs.com联系，而不是使用问题跟踪器。

致谢

许可证

MIT许可证（MIT）。有关更多信息，请参阅许可证文件。

jrodella / laravel-sqs-fifo-queue

维护者

详细信息