php - 如何在 Laravel 队列中容纳 Amazon FIFO SQS？

Question

亚马逊宣布了他们的 new FIFO SQS service我想在 Laravel Queue 中使用它来解决一些并发问题。

我创建了几个新队列并更改了配置。但是，我收到了一个 MissingParameter 错误，上面写着

The request must contain the parameter MessageGroupId.

所以我修改了文件vendor/laravel/framework/src/Illuminate/Queue/SqsQueue.php

public function pushRaw($payload, $queue = null, array $options = [])
{
    $response = $this->sqs->sendMessage(['QueueUrl' => $this->getQueue($queue), 'MessageBody' => $payload,
        'MessageGroupId' => env('APP_ENV', getenv('APP_ENV'))]);

    return $response->get('MessageId');
}

public function later($delay, $job, $data = '', $queue = null)
{
    $payload = $this->createPayload($job, $data);

    $delay = $this->getSeconds($delay);

    return $this->sqs->sendMessage([
        'QueueUrl' => $this->getQueue($queue), 'MessageBody' => $payload, 'DelaySeconds' => $delay,
        'MessageGroupId' => env('APP_ENV', getenv('APP_ENV'))
    ])->get('MessageId');
}

我使用 APP_ENV 作为组 ID(这是一个单一的消息队列，所以实际上它并不重要。我只希望一切都是先进先出)。

但我仍然收到相同的错误消息。我该如何解决？任何帮助将不胜感激。

(顺便说一句，SDK在哪里定义了sendMessage？我可以找到它的 stub ，但我没有找到详细的实现)

Answer 1

我想向可能遇到同样问题的其他人指出，尽管编辑 SqsQueue.php 有效，但它很容易被 composer install 或 Composer 更新。另一种方法是为 SQS FIFO 实现一个新的 Illuminate\Queue\Connectors\ConnectorInterface，然后将其添加到 Laravel 的队列管理器。

我的做法是这样的:

1. 创建一个扩展 Illuminate\Queue\SqsQueue 但支持 SQS FIFO 的新 SqsFifoQueue 类。
2. 创建一个扩展 Illuminate\Queue\Connectors\SqsConnector 的新 SqsFifoConnector 类，它将使用 SqsFifoQueue 建立连接。
3. 创建一个新的 SqsFifoServiceProvider，将 SqsFifoConnector 注册到 Laravel 的队列管理器。
4. 将 SqsFifoServiceProvider 添加到您的 config/app.php。
5. 更新 config/queue.php 以使用新的 SQS FIFO 队列驱动程序。

示例:

1. 创建一个扩展 Illuminate\Queue\SqsQueue 但支持 SQS FIFO 的新 SqsFifoQueue 类。

   <?php
   
   class SqsFifoQueue extends \Illuminate\Queue\SqsQueue
   {
       public function pushRaw($payload, $queue = null, array $options = [])
       {
           $response = $this->sqs->sendMessage([
               'QueueUrl' => $this->getQueue($queue),
               'MessageBody' => $payload,
               'MessageGroupId' => uniqid(),
               'MessageDeduplicationId' => uniqid(),
           ]);
   
           return $response->get('MessageId');
       }
   }
   

2. 创建一个扩展 Illuminate\Queue\Connectors\SqsConnector 的新 SqsFifoConnector 类，它将使用 SqsFifoQueue 建立连接。

   <?php
   
   use Aws\Sqs\SqsClient;
   use Illuminate\Support\Arr;
   
   class SqsFifoConnector extends \Illuminate\Queue\Connectors\SqsConnector
   {
       public function connect(array $config)
       {
           $config = $this->getDefaultConfiguration($config);
   
           if ($config['key'] && $config['secret']) {
               $config['credentials'] = Arr::only($config, ['key', 'secret']);
           }
   
           return new SqsFifoQueue(
               new SqsClient($config), $config['queue'], Arr::get($config, 'prefix', '')
           );
       }
   }
   

3. 创建一个新的 SqsFifoServiceProvider，将 SqsFifoConnector 注册到 Laravel 的队列管理器。

   <?php
   
   class SqsFifoServiceProvider extends \Illuminate\Support\ServiceProvider
   {
       public function register()
       {
           $this->app->afterResolving('queue', function ($manager) {
               $manager->addConnector('sqsfifo', function () {
                   return new SqsFifoConnector;
               });
           });
       }
   }
   

4. 将 SqsFifoServiceProvider 添加到您的 config/app.php。

   <?php
   
   return [
       'providers'     => [
           ...
           SqsFifoServiceProvider::class,
       ],
   ];
   

5. 更新 config/queue.php 以使用新的 SQS FIFO 队列驱动程序。

   <?php
   
   return [
   
       'default' => 'sqsfifo',
   
       'connections' => [
           'sqsfifo' => [
               'driver' => 'sqsfifo',
               'key'    => 'my_key'
               'secret' => 'my_secret',
               'queue'  => 'my_queue_url',
               'region' => 'my_sqs_region',
           ],
       ],
   ];
   

那么您的队列现在应该支持 SQS FIFO 队列。

无耻插件:在执行上述步骤时，我创建了一个 laravel-sqs-fifo在 https://github.com/maqe/laravel-sqs-fifo 处处理此问题的 Composer 包.