中间件

以 Markdown 格式查看在 GitHub 上编辑文档位置 2cc7b2b

中间件会以洋葱式管道包装作业执行。每个中间件都可以在作业执行前后运行逻辑、短路执行或捕获异常。Queuety 内置了五种中间件，并支持通过 Contracts\Middleware 接口创建自定义中间件。

中间件适用于可分发的作业类（实现 Contracts\Job 并使用 Dispatchable 特征的类）。

`Contracts\Middleware` 接口

PHP

use Queuety\Contracts\Middleware;

interface Middleware {
    public function handle( object $job, \Closure $next ): void;
}

$job 参数是正在处理的作业实例。调用 $next( $job ) 将执行传递给管道中的下一个中间件（如果这是最后一个中间件，则传递给作业的 handle() 方法）。

为作业添加中间件

在作业类中定义一个 middleware() 方法，返回中间件实例数组：

PHP

use Queuety\Contracts\Job;
use Queuety\Dispatchable;
use Queuety\Middleware\RateLimited;
use Queuety\Middleware\Timeout;

class SyncInventoryJob implements Job {
    use Dispatchable;

    public function __construct(
        public readonly string $sku,
    ) {}

    public function handle(): void {
        // Sync inventory for this SKU...
    }

    public function middleware(): array {
        return [
            new Timeout( seconds: 30 ),
            new RateLimited( max: 100, window: 60 ),
        ];
    }
}

管道执行顺序

中间件按数组顺序应用，第一个中间件位于最外层。如果你的 middleware() 方法返回 [A, B, C]，执行顺序如下：

代码

A::handle() -> B::handle() -> C::handle() -> $job->handle()

每个中间件都会包装下一个。A 在作业执行前最先运行，并在作业执行后最后运行，因此非常适合用于计时或 try/catch 包装。

编写自定义中间件

实现 Contracts\Middleware 接口：

PHP

use Queuety\Contracts\Middleware;

class LogExecution implements Middleware {
    public function handle( object $job, \Closure $next ): void {
        $class = get_class( $job );
        error_log( "[Queuety] Starting: {$class}" );

        $start = microtime( true );

        try {
            $next( $job );
        } finally {
            $elapsed = round( microtime( true ) - $start, 3 );
            error_log( "[Queuety] Finished: {$class} in {$elapsed}s" );
        }
    }
}

在作业中使用它：

PHP

public function middleware(): array {
    return [
        new LogExecution(),
    ];
}

短路执行

中间件可以不调用 $next()，从而完全跳过作业：

PHP

class SkipOnMaintenanceMode implements Middleware {
    public function handle( object $job, \Closure $next ): void {
        if ( wp_is_maintenance_mode() ) {
            return; // Job silently skipped.
        }

        $next( $job );
    }
}

内置中间件

`ThrottlesExceptions`

限制作业处理器抛出异常的次数，以防止作业风暴。如果某个处理器在一个时间窗口内抛出异常过多，该处理器后续的作业会被拒绝并收到 RuntimeException，而不会被处理。这样可以防止故障的外部服务一次耗尽许多作业的所有重试次数。

PHP

use Queuety\Middleware\ThrottlesExceptions;

public function middleware(): array {
    return [
        new ThrottlesExceptions( max_attempts: 10, decay_minutes: 10 ),
    ];
}

参数	类型	默认值	描述
`$max_attempts`	`int`	`10`	衰减窗口内允许的最大异常数
`$decay_minutes`	`int`	`10`	窗口时长（分钟）

异常计数按处理器类记录在 Queuety 的锁表中，并由所有工作进程共享。达到阈值时，中间件会抛出 RuntimeException，而不调用作业；工作进程会像处理其他故障一样处理它：该次尝试会计入 max_attempts，作业会按常规退避策略重试，尝试次数用尽后则会被存入失败队列。

例如：如果外部 API 宕机，前 10 次失败会正常处理（每次都会增加异常计数）。此后，尝试会快速失败，不再调用 API，直到连续 10 分钟没有新异常为止；每记录一次异常，窗口就会重新开始。

`RateLimited`

在一个时间窗口内强制限制执行次数。如果在中间件中超出限制，就会抛出 RateLimitExceededException；工作进程会将其作为普通失败尝试处理，按常规退避策略重试，并计入 max_attempts。中间件向工作进程的速率限制器注册限制后，在达到限制期间，后续作业也会在运行前被放回 pending（参见速率限制）。

PHP

use Queuety\Middleware\RateLimited;

public function middleware(): array {
    return [
        new RateLimited( max: 60, window: 60 ), // 60 per minute
    ];
}

参数	类型	描述
`$max`	`int`	窗口内允许的最大执行次数
`$window`	`int`	窗口时长（秒）

速率限制器使用 Queuety 的内部 RateLimiter 实例，因此限制会在所有工作进程间生效。

`Timeout`

使用 pcntl_alarm 为每个作业设置执行超时，并计划在作业运行时间过长时抛出 TimeoutException。目前超时并未生效（#10）：Queuety 未启用异步信号，因此作业执行期间不会运行闹钟处理程序。中间件返回时还会重置闹钟，从而取消工作进程自身的作业超时闹钟。

PHP

use Queuety\Middleware\Timeout;

public function middleware(): array {
    return [
        new Timeout( seconds: 30 ),
    ];
}

参数	类型	描述
`$seconds`	`int`	最大执行时间（秒）

需要安装 pcntl PHP 扩展。如果没有 pcntl，中间件会静默跳过（不会强制执行超时）。

`UniqueJob`

防止具有相同键的作业并发执行。使用数据库锁表确保同一时间只有一个实例运行。如果锁已被占用，作业会被静默跳过（不会重试）。

PHP

use Queuety\Middleware\UniqueJob;

public function middleware(): array {
    return [
        new UniqueJob(),                         // Lock key defaults to the job class name
        new UniqueJob( key: 'sync_inventory' ),   // Custom lock key
    ];
}

参数	类型	默认值	描述
`$key`	`string\|null`	`null`（使用类名）	唯一锁键

`WithoutOverlapping`

防止共享逻辑键的作业重叠执行，并自动过期锁，以免工作进程崩溃后留下死锁。与 UniqueJob 类似，但专为长时间运行的操作设计，锁会在超时后自动释放。

PHP

use Queuety\Middleware\WithoutOverlapping;

public function middleware(): array {
    return [
        new WithoutOverlapping( key: 'import_products', release_after: 600 ),
    ];
}

参数	类型	默认值	描述
`$key`	`string`	（必填）	逻辑操作键
`$release_after`	`int`	`300`	自动释放前的最长锁定时间（秒）

如果锁已被占用，作业会被静默跳过。每次尝试获取锁之前，都会自动清理过期锁。

示例：组合使用中间件

PHP

use Queuety\Contracts\Job;
use Queuety\Dispatchable;
use Queuety\Middleware\RateLimited;
use Queuety\Middleware\Timeout;
use Queuety\Middleware\WithoutOverlapping;

class ImportProductsJob implements Job {
    use Dispatchable;

    public function __construct(
        public readonly int $supplier_id,
    ) {}

    public function handle(): void {
        // Fetch and import products from the supplier...
    }

    public function middleware(): array {
        return [
            new Timeout( seconds: 300 ),
            new WithoutOverlapping(
                key: "import_supplier_{$this->supplier_id}",
                release_after: 600,
            ),
            new RateLimited( max: 5, window: 60 ),
        ];
    }
}

此作业将会：

运行超过 5 分钟时被终止（Timeout）
如果同一供应商的另一个导入任务正在运行，则跳过（WithoutOverlapping）
对所有供应商合计，每分钟限制为 5 次导入（RateLimited）

中间件

Contracts\Middleware 接口