Laravel 11 Shopify API Job限流失效及全失败通知需求

Laravel 11队列支持限流，我的应用需要调用Shopify API完成商品拉取、订单备注添加、物流追踪号录入等操作。Shopify基础版API限制为每秒最多2次请求、每分钟最多40次请求。

我编写了通用ShopifyApiRequestJob类处理所有Shopify API调用，并在AppServiceProvider.php中定义了限流规则，但测试时未达到预期效果：分发10个Job后仅3个成功，其余7个因MaxAttemptsExceededException失败。

限流规则配置（AppServiceProvider.php） #

public function boot(): void
{
    // Rate limit Shopify API requests set to 2 per second and 40 per minute
    RateLimiter::for('shopify-api-requests', function (object $job) {
        return [
            Limit::perSecond(2),
            Limit::perMinute(40),
        ];
    });
}

通用Job类代码（ShopifyApiRequestJob.php） #

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

    public $endpoint;
    public $method;
    public $data;
    
    public function __construct(string $endpoint, string $method = 'GET', array $data = null)
    {
        $this->endpoint = $endpoint;
        $this->method = $method;
        $this->data = $data;
    }
    
    public function backoff(): array
    {
        return [1, 5, 10];
    }
    
    public function tries(): int
    {
        return 3;
    }
    
    public function middleware(): array
    {
        return [
            new RateLimited('shopify-api-requests'),
            //new WithoutOverlapping('shopify-api-requests')
        ];
    }
    
    public function handle()
    {
        // Construct the full URL
        $url = 'https://' . config('settings.SHOPIFY_API_DOMAIN') . '/admin/api/' . config('settings.SHOPIFY_API_VERSION') . '/' . $this->endpoint;

        $response = Http::withHeaders([
            'X-Shopify-Access-Token' => config('settings.SHOPIFY_API_KEY'),
            'Content-Type' => 'application/json',
        ])->{$this->method}($url, $this->data);

        // Handle the response as needed (e.g., log it, store it, etc.)
        if ($response->failed()) {
            // Handle failure (e.g., retry the job, log the error, etc.)
            Log::error("Shopify API Request Failed (" . $response->status() . "): " . $response->body() . " " . $url);
        } else {
            // Handle success (e.g., process the response, store it, etc.)
            Log::info("Shopify API Request Successful");
        }
    }
}

测试日志 #

[2024-08-24 19:29:11] local.INFO: Shopify API Request Successful  
[2024-08-24 19:29:17] local.INFO: Shopify API Request Successful  
[2024-08-24 19:29:20] local.INFO: Shopify API Request Successful  
[2024-08-24 19:29:26] local.ERROR: App\Jobs\ShopifyApiRequestJob has been attempted too many times. {"exception":"[object] (Illuminate\\Queue\\MaxAttemptsExceededException(code: 0): [...]
[2024-08-24 19:29:26] local.ERROR: App\Jobs\ShopifyApiRequestJob has been attempted too many times. {"exception":"[object] (Illuminate\\Queue\\MaxAttemptsExceededException(code: 0): [...]
[2024-08-24 19:29:26] local.ERROR: App\Jobs\ShopifyApiRequestJob has been attempted too many times. {"exception":"[object] (Illuminate\\Queue\\MaxAttemptsExceededException(code: 0): [...]
[2024-08-24 19:29:26] local.ERROR: App\Jobs\ShopifyApiRequestJob has been attempted too many times. {"exception":"[object] (Illuminate\\Queue\\MaxAttemptsExceededException(code: 0): [...]
[2024-08-24 19:29:26] local.ERROR: App\Jobs\ShopifyApiRequestJob has been attempted too many times. {"exception":"[object] (Illuminate\\Queue\\MaxAttemptsExceededException(code: 0): [...]
[2024-08-24 19:29:26] local.ERROR: App\Jobs\ShopifyApiRequestJob has been attempted too many times. {"exception":"[object] (Illuminate\\Queue\\MaxAttemptsExceededException(code: 0): [...]
[2024-08-24 19:29:26] local.ERROR: App\Jobs\ShopifyApiRequestJob has been attempted too many times. {"exception":"[object] (Illuminate\\Queue\\MaxAttemptsExceededException(code: 0): [...]

需求：

1. 排查Job限流配置错误
2. 实现仅在Job所有重试失败时接收通知，而非每次重试失败都通知

一、限流配置错误排查 #

1. 为限流规则添加唯一标识 #

当前限流规则未绑定店铺唯一标识，导致限流逻辑未按预期分组生效。修改规则，用店铺域名作为限流键，确保每个店铺独立计算请求次数：

use Illuminate\Cache\RateLimiting\Limit;
use Illuminate\Support\Facades\RateLimiter;

public function boot(): void
{
    RateLimiter::for('shopify-api-requests', function (object $job) {
        $key = config('settings.SHOPIFY_API_DOMAIN');
        
        return [
            Limit::perSecond(2)->by($key),
            Limit::perMinute(40)->by($key),
        ];
    });
}

2. 修正重试触发逻辑 #

Laravel队列仅在handle方法抛出异常时才会将Job放回队列重试。当前代码仅记录日志未抛出异常，导致失败的Job直接终止，无法触发重试。修改handle方法：

public function handle()
{
    $url = 'https://' . config('settings.SHOPIFY_API_DOMAIN') . '/admin/api/' . config('settings.SHOPIFY_API_VERSION') . '/' . $this->endpoint;

    $response = Http::withHeaders([
        'X-Shopify-Access-Token' => config('settings.SHOPIFY_API_KEY'),
        'Content-Type' => 'application/json',
    ])->{$this->method}($url, $this->data);

    if ($response->failed()) {
        $errorMsg = "Shopify API Request Failed (" . $response->status() . "): " . $response->body() . " " . $url;
        Log::error($errorMsg);
        // 抛出异常触发重试
        throw new \Exception($errorMsg);
    } else {
        Log::info("Shopify API Request Successful");
    }
}

3. 调整退避与重试参数 #

当前退避时间总和较短，可能导致在重试次数内无法等待限流释放。可以延长退避时间或增加重试次数：

public function backoff(): array
{
    // 递增退避，给限流足够恢复时间
    return [2, 8, 15, 30];
}

public function tries(): int
{
    return 4;
}

4. 检查队列Worker配置 #

确保队列Worker未通过--tries参数覆盖Job的重试配置，同时限制Worker并发数避免过载：

php artisan queue:work --tries=4 --max-jobs=100

二、仅在所有重试失败时发送通知 #

利用Laravel队列的failed方法，该方法仅在Job所有重试耗尽后触发：

1. 在Job类中添加failed方法 #

use Illuminate\Notifications\Notifiable;
use App\Notifications\ShopifyJobFailedNotification;

class ShopifyApiRequestJob implements ShouldQueue
{
    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels, Notifiable;

    // ... 其他代码 ...

    /**
     * 所有重试失败后触发
     */
    public function failed(\Throwable $exception)
    {
        $this->notify(new ShopifyJobFailedNotification($this, $exception));
    }
}

2. 创建失败通知类 #

namespace App\Notifications;

use Illuminate\Bus\Queueable;
use Illuminate\Notifications\Notification;
use Illuminate\Notifications\Messages\MailMessage;

class ShopifyJobFailedNotification extends Notification
{
    use Queueable;

    protected $job;
    protected $exception;

    public function __construct($job, $exception)
    {
        $this->job = $job;
        $this->exception = $exception;
    }

    public function via($notifiable)
    {
        return ['mail']; // 可按需切换为Slack、短信等渠道
    }

    public function toMail($notifiable)
    {
        return (new MailMessage)
            ->subject('Shopify API Job 执行失败')
            ->line("失败Endpoint: {$this->job->endpoint}")
            ->line("错误信息: {$this->exception->getMessage()}")
            ->line("失败时间: " . now()->toDateTimeString());
    }
}

3. 指定通知接收人 #

在Job类中添加方法配置接收邮箱：

public function routeNotificationForMail()
{
    return 'admin@yourdomain.com';
}

内容的提问来源于stack exchange，提问作者Jan