php - 将 SQS 与多个 Laravel 队列读取器一起使用时出错

Question

我正在使用 Laravel Jobs 从 SQS 队列中读取消息(Laravel 5.7 版)

关注 Laravel indications我正在使用 supervisor 同时运行多个 queue:work 进程。

一切顺利，直到我收到与消息可用性相关的 SQS 错误:

InvalidParameterValue (client): Value 
... for parameter ReceiptHandle is invalid. Reason: Message does not exist or 
is not available for visibility timeout change. - <?xml version="1.0"?> 
<ErrorResponse xmlns="http://queue.amazonaws.com/doc/2012-11-05/"><Error>
<Type>Sender</Type><Code>InvalidParameterValue</Code><Message>Value ...
for parameter ReceiptHandle is invalid. Reason: Message does not exist or is 
not available for visibility timeout change.</Message><Detail/></Error>
<RequestId>8c1d28b7-a02c-5059-8b65-7c6292a0e56e</RequestId></ErrorResponse> 
{"exception":"[object] (Aws\\Sqs\\Exception\\SqsException(code: 0): Error 
executing \"ChangeMessageVisibility\" on \"https://sqs.eu-central-
1.amazonaws.com/123123123123/myQueue\"; AWS HTTP error: Client error: `POST 
https://sqs.eu-central-1.amazonaws.com/123123123123/myQueue` resulted in a 
`400 Bad Request` response:

特别奇怪的是Message does not exist or is not available for visibility timeout change.

每个主管进程调用 command=php/home/application/artisan queue:work 没有 --sleep=3 (我希望进程是响应式(Reactive)的并且不等待 3 秒以防队列中没有任何内容)或 --tries=3(我需要完成所有任务，所以我不限制 尝试参数)

如果消息不存在(我不能排除这种可能性)，为什么进程要从队列中获取它？我能做些什么来阻止它吗？

Answer 1

我也在生产中间歇性地看到过这个错误，我们为单个 SQS 队列运行了大量的消费者。在我们的例子中，我非常确信错误是由于 SQS 的 at-least-once 引起的。交付语义。本质上，在极少数情况下，一条消息可以传递两次或更多次。

Laravel 的队列 worker 命令不是严格幂等的，因为它会在尝试释放或删除不再可用的 SQS 消息时抛出异常(即，因为它已被另一个队列 worker 进程删除，该进程接收到重复的来自 SQS 的消息)。

我们的解决方法是尝试检测何时收到重复的消息，然后尝试安全地将消息释放回队列。如果当前正在处理该消息的其他队列 worker 成功，它将删除该消息，并且不会再次接收它。如果另一个队列 worker 失败，则消息将被释放并稍后再次接收。像这样:

<?php

use Aws\Sqs\Exception\SqsException;
use Illuminate\Bus\Queueable;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Bus\Dispatchable;
use Illuminate\Queue\InteractsWithQueue;
use Illuminate\Queue\SerializesModels;
use Illuminate\Support\Facades\Cache;
use Illuminate\Support\Str;

class ProcessPodcast implements ShouldQueue
{
  use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;

  private $jobId;

  public function __construct($jobId)
  {
    $this->jobId = $jobId;
  }

  public function handle()
  {
    $acquired = Cache::lock("process-podcast-$this->jobId")->get(function () {
      // Process the podcast (NB: this should be idempotent)
    });

    if (!$acquired) {
      $this->releaseDuplicateMessage($delay = 60);
    }
  }

  private function releaseDuplicateMessage($delay)
  {
    try {
      $this->release($delay);
    } catch (Exception $ex) {
      if (!$this->causedByMessageNoLongerAvailable($ex)) {
        throw $ex;
      }
    }
  }

  private function causedByMessageNoLongerAvailable(Exception $ex): bool
  {
    return $ex instanceof SqsException &&
           Str::contains(
             $ex->getAwsErrorMessage(),
             "Message does not exist or is not available for visibility timeout change"
           );
  }
}