README

一个现代化、高性能的 PHP 队列客户端，支持多种队列驱动，专为 PHP 8.1+ 设计，完美支持 PHP 8.5 管道操作符新特性。

目录

• 特性
• 安装
• 环境要求
• 驱动依赖
• 快速开始
• 驱动配置
• 核心功能
• 中间件
• 上下文管理
• 工具类
• PHP 8.5 管道操作符
• API 参考
• 项目结构
• 测试

特性

• 多驱动支持（Redis、Beanstalkd、AMQP、Kafka、Database、Sync）
• 自动驱动检测与配置
• 统一的队列操作 API
• 支持延迟入队功能
• 支持批量操作（批量入队、批量出队）
• 支持队列监控功能
• 完善的异常处理机制
• 支持事务功能
• 中间件支持（重试、超时、日志等）
• 上下文管理（Context）
• 工具类支持（QueueUtil）
• 符合 PSR 规范（PSR-1、PSR-2、PSR-4、PSR-12）
• PHP 8.1+ 兼容，完美支持 PHP 8.5 管道操作符

安装

composer require kode/queue

环境要求

驱动依赖

快速开始

基本使用

use Kode\Queue\Factory;

// 创建队列实例
$queue = Factory::create([
    'default' => 'redis',
    'connections' => [
        'redis' => [
            'host' => '127.0.0.1',
            'port' => 6379,
        ],
    ],
]);

// 入队
$jobId = $queue->push('SendEmail', ['user_id' => 123]);

// 延迟入队（60秒后执行）
$jobId = $queue->later(60, 'SendEmail', ['user_id' => 123]);

// 批量入队
$jobIds = $queue->bulk(['SendEmail', 'ProcessOrder'], ['user_id' => 123]);

// 出队
$job = $queue->pop();
if ($job) {
    // 处理任务
    echo $job['job']; // 任务名称
    print_r($job['data']); // 任务数据
}

// 查看队列长度
$size = $queue->size();

// 查看队列统计信息
$stats = $queue->stats();

使用指定驱动

use Kode\Queue\Factory;

// 使用 Sync 驱动（同步执行，用于测试）
$queue = Factory::createWithDriver('sync');

// 使用 Redis 驱动
$queue = Factory::createWithDriver('redis', [
    'host' => '127.0.0.1',
    'port' => 6379,
    'database' => 0,
    'password' => null,
]);

// 使用 Database 驱动
$queue = Factory::createWithDriver('database', [
    'dsn' => 'mysql:host=127.0.0.1;dbname=queue',
    'username' => 'root',
    'password' => '',
    'table' => 'jobs',
]);

// 使用 Beanstalkd 驱动
$queue = Factory::createWithDriver('beanstalkd', [
    'host' => '127.0.0.1',
    'port' => 11300,
    'tube' => 'default',
]);

// 使用 AMQP 驱动
$queue = Factory::createWithDriver('amqp', [
    'host' => '127.0.0.1',
    'port' => 5672,
    'username' => 'guest',
    'password' => 'guest',
    'vhost' => '/',
    'queue' => 'default',
]);

// 使用 Kafka 驱动
$queue = Factory::createWithDriver('kafka', [
    'bootstrap_servers' => '127.0.0.1:9092',
    'topic' => 'queue',
    'group_id' => 'queue-consumer',
]);

驱动配置

完整配置示例

$config = [
    'default' => 'redis',
    'connections' => [
        'redis' => [
            'driver' => 'redis',
            'host' => '127.0.0.1',
            'port' => 6379,
            'database' => 0,
            'password' => null,
            'options' => [],
        ],
        'beanstalkd' => [
            'driver' => 'beanstalkd',
            'host' => '127.0.0.1',
            'port' => 11300,
            'tube' => 'default',
        ],
        'amqp' => [
            'driver' => 'amqp',
            'host' => '127.0.0.1',
            'port' => 5672,
            'username' => 'guest',
            'password' => 'guest',
            'vhost' => '/',
            'queue' => 'default',
        ],
        'kafka' => [
            'driver' => 'kafka',
            'bootstrap_servers' => '127.0.0.1:9092',
            'topic' => 'queue',
            'group_id' => 'queue-consumer',
        ],
        'sync' => [
            'driver' => 'sync',
        ],
        'database' => [
            'driver' => 'database',
            'dsn' => 'mysql:host=127.0.0.1;dbname=queue',
            'username' => 'root',
            'password' => '',
            'table' => 'jobs',
        ],
    ],
];

$queue = Factory::create($config);

各驱动配置说明

Redis 驱动

Database 驱动

Beanstalkd 驱动

AMQP 驱动

Kafka 驱动

核心功能

全局队列和局部队列

use Kode\Queue\Factory;

$queue = Factory::createWithDriver('sync');

// 获取全局队列实例（单例模式）
$globalQueue1 = $queue->global('orders');
$globalQueue2 = $queue->global('orders');
// $globalQueue1 和 $globalQueue2 是同一个实例

// 全局队列应用场景：
// - 系统级任务队列
// - 跨模块共享的队列
// - 需要在多个地方访问的队列

// 获取局部队列实例（每次创建新实例）
$localQueue1 = $queue->local('emails');
$localQueue2 = $queue->local('emails');
// $localQueue1 和 $localQueue2 是不同实例

// 局部队列应用场景：
// - 临时任务队列
// - 单次操作的队列
// - 不需要跨模块共享的队列

使用事务

use Kode\Queue\Factory;

$queue = Factory::createWithDriver('database');

try {
    $queue->beginTransaction();
    
    // 执行多个队列操作
    $jobId1 = $queue->push('Task1', ['data' => 'value1']);
    $jobId2 = $queue->push('Task2', ['data' => 'value2']);
    
    $queue->commit();
} catch (\Exception $e) {
    $queue->rollback();
    // 处理异常
}

延迟任务

use Kode\Queue\Factory;

$queue = Factory::createWithDriver('sync');

// 延迟 60 秒执行
$jobId = $queue->later(60, 'SendEmail', ['user_id' => 123]);

// 延迟 5 分钟执行
$jobId = $queue->later(300, 'ProcessOrder', ['order_id' => 456]);

// 延迟 1 小时执行
$jobId = $queue->later(3600, 'CleanupTask', ['type' => 'logs']);

批量操作

use Kode\Queue\Factory;

$queue = Factory::createWithDriver('sync');

// 批量推送任务
$jobIds = $queue->bulk([
    'SendEmail',
    'SendSms',
    'SendPush',
], ['user_id' => 123]);

// 批量推送不同数据的任务
$jobs = [
    ['job' => 'SendEmail', 'data' => ['user_id' => 1]],
    ['job' => 'SendEmail', 'data' => ['user_id' => 2]],
    ['job' => 'SendEmail', 'data' => ['user_id' => 3]],
];

foreach ($jobs as $job) {
    $queue->push($job['job'], $job['data']);
}

中间件

内置中间件

使用中间件

use Kode\Queue\Factory;
use Kode\Queue\Middleware\RetryMiddleware;
use Kode\Queue\Middleware\LogMiddleware;
use Kode\Queue\Middleware\RateLimitMiddleware;

$queue = Factory::createWithDriver('sync');

// 添加中间件
$queue->addMiddleware(new RetryMiddleware(3, 100, 2.0))
      ->addMiddleware(new LogMiddleware())
      ->addMiddleware(new RateLimitMiddleware(10, 1));

// 使用队列
$jobId = $queue->push('SendEmail', ['user_id' => 123]);

自定义中间件

use Kode\Queue\Middleware\MiddlewareInterface;

class CustomMiddleware implements MiddlewareInterface {
    public function handle(callable $next, string $method, array $parameters) {
        // 前置处理
        echo "Before: $method\n";
        
        // 执行下一个处理器
        $result = $next($parameters);
        
        // 后置处理
        echo "After: $method\n";
        
        return $result;
    }
}

// 使用自定义中间件
$queue->addMiddleware(new CustomMiddleware());

上下文管理

Context 类用于封装和管理队列任务的上下文信息，提供便捷的方法来访问和操作任务数据。

创建上下文

use Kode\Queue\Context\Context;

// 从任务数据创建上下文
$context = new Context([
    'id' => 'job_123',
    'job' => 'SendEmail',
    'data' => ['user_id' => 123],
    'attempts' => 0,
    'created_at' => time(),
]);

// 从数组创建
$context = Context::fromArray($jobData);

// 从 JSON 创建
$context = Context::fromJson('{"id":"job_123","job":"SendEmail",...}');

获取任务信息

// 获取任务ID
$jobId = $context->getJobId();        // 'job_123'

// 获取任务名称
$jobName = $context->getJob();        // 'SendEmail'

// 获取任务数据
$jobData = $context->getData();       // ['user_id' => 123]

// 获取重试次数
$attempts = $context->getAttempts();  // 0

// 获取创建时间
$createdAt = $context->getCreatedAt(); // 时间戳

// 获取队列名称
$queue = $context->getQueue();        // 队列名称

// 获取完整负载
$payload = $context->getPayload();    // 完整数组数据

操作上下文

// 增加重试次数
$context->incrementAttempts();

// 转换为数组
$payload = $context->toArray();

// 转换为 JSON
$json = $context->toJson();

在任务处理器中使用

use Kode\Queue\Context\Context;

class SendEmailHandler {
    public function handle(array $jobData) {
        $context = new Context($jobData);
        
        // 获取任务信息
        $userId = $context->getData()['user_id'];
        $attempts = $context->getAttempts();
        
        // 执行任务逻辑
        try {
            $this->sendEmail($userId);
        } catch (\Exception $e) {
            // 记录重试
            $context->incrementAttempts();
            
            // 判断是否超过最大重试次数
            if ($context->getAttempts() >= 3) {
                throw $e;
            }
            
            // 重新入队
            return false;
        }
        
        return true;
    }
}

工具类

QueueUtil 类提供队列操作中常用的工具方法。

生成任务ID

use Kode\Queue\Util\QueueUtil;

// 生成唯一任务ID
$jobId = QueueUtil::generateJobId();
// 输出: 67e8f12345678.12345678

创建和解析负载

use Kode\Queue\Util\QueueUtil;

// 创建任务负载
$payload = QueueUtil::createPayload('SendEmail', ['user_id' => 123]);
// 输出: {"job":"SendEmail","data":{"user_id":123},"id":"...","attempts":0,"created_at":1234567890}

// 解析任务负载
$data = QueueUtil::parsePayload($payload);
// 输出: ['job' => 'SendEmail', 'data' => ['user_id' => 123], ...]

检查任务状态

use Kode\Queue\Util\QueueUtil;

// 检查任务是否就绪（用于延迟任务）
$job = ['available_at' => time() - 10];
$isReady = QueueUtil::isJobReady($job);  // true

$job = ['available_at' => time() + 60];
$isReady = QueueUtil::isJobReady($job);  // false

计算延迟

use Kode\Queue\Util\QueueUtil;

// 计算延迟后的时间戳
$timestamp = QueueUtil::calculateDelay(60);  // 当前时间 + 60秒

格式化队列大小

use Kode\Queue\Util\QueueUtil;

// 格式化队列大小
echo QueueUtil::formatSize(500);     // "500"
echo QueueUtil::formatSize(1500);    // "1.50K"
echo QueueUtil::formatSize(1500000); // "1.50M"

验证名称

use Kode\Queue\Util\QueueUtil;

// 验证队列名称
QueueUtil::validateQueueName('my-queue');    // true
QueueUtil::validateQueueName('my_queue');    // true
QueueUtil::validateQueueName('my.queue');    // true
QueueUtil::validateQueueName('my queue');    // false (包含空格)

// 验证任务名称
QueueUtil::validateJobName('SendEmail');         // true
QueueUtil::validateJobName('App\Jobs\SendEmail'); // true
QueueUtil::validateJobName('123SendEmail');      // false (数字开头)

指数退避

use Kode\Queue\Util\QueueUtil;

// 计算指数退避延迟时间（毫秒）
QueueUtil::exponentialBackoff(1, 100, 2.0);  // 100ms (第1次重试)
QueueUtil::exponentialBackoff(2, 100, 2.0);  // 200ms (第2次重试)
QueueUtil::exponentialBackoff(3, 100, 2.0);  // 400ms (第3次重试)
QueueUtil::exponentialBackoff(4, 100, 2.0);  // 800ms (第4次重试)

// 自定义参数
QueueUtil::exponentialBackoff(3, 50, 1.5);   // 112ms (基础延迟50ms，乘数1.5)

PHP 8.5 管道操作符支持

PHP 8.5 引入了管道操作符 |>，本包完美支持这一新特性。

传统写法

use Kode\Queue\Factory;
use Kode\Queue\Middleware\LogMiddleware;

$queue = Factory::createWithDriver('sync');
$queue->addMiddleware(new LogMiddleware());
$jobId = $queue->push('SendEmail', ['user_id' => 123]);

PHP 8.5 管道操作符写法

use Kode\Queue\Factory;
use Kode\Queue\Middleware\LogMiddleware;
use Kode\Queue\Middleware\RetryMiddleware;

// 基本管道操作
$jobId = Factory::createWithDriver('sync')
    |> fn($q) => $q->push('SendEmail', ['user_id' => 123]);

// 管道操作与中间件
$jobId = Factory::createWithDriver('sync')
    |> fn($q) => $q->addMiddleware(new LogMiddleware())
    |> fn($q) => $q->addMiddleware(new RetryMiddleware(3, 100, 2.0))
    |> fn($q) => $q->push('SendEmail', ['user_id' => 123]);

// 管道操作与全局队列
$jobId = Factory::createWithDriver('sync')
    |> fn($q) => $q->global('emails')
    |> fn($q) => $q->push('SendEmail', ['user_id' => 123]);

// 管道操作与延迟任务
$jobId = Factory::createWithDriver('sync')
    |> fn($q) => $q->later(60, 'DelayedJob', ['data' => 'delayed']);

// 管道操作与批量任务
$jobIds = Factory::createWithDriver('sync')
    |> fn($q) => $q->bulk(['Job1', 'Job2', 'Job3'], ['data' => 'bulk']);

兼容 PHP 8.1-8.4 的管道方法

对于 PHP 8.5 以下版本，可以使用 pipe() 方法实现类似功能：

use Kode\Queue\Factory;
use Kode\Queue\Middleware\LogMiddleware;

// 使用 pipe 方法
$jobId = Factory::createWithDriver('sync')
    ->pipe(fn($q) => $q->addMiddleware(new LogMiddleware()))
    ->pipe(fn($q) => $q->push('SendEmail', ['user_id' => 123]));

// 链式 pipe 调用
$result = Factory::createWithDriver('sync')
    ->pipe(fn($q) => $q->addMiddleware(new LogMiddleware()))
    ->pipe(fn($q) => $q->push('Task1', ['data' => 'value1']))
    ->pipe(fn($q) => $q->push('Task2', ['data' => 'value2']))
    ->pipe(fn($q) => $q->size());

API 参考

QueueInterface

项目结构

src/
├── Driver/                  # 驱动目录
│   ├── DriverInterface.php  # 驱动接口
│   ├── SyncDriver.php       # 同步驱动
│   ├── DatabaseDriver.php   # 数据库驱动
│   ├── RedisDriver.php      # Redis 驱动
│   ├── BeanstalkdDriver.php # Beanstalkd 驱动
│   ├── AmqpDriver.php       # AMQP 驱动
│   └── KafkaDriver.php      # Kafka 驱动
├── Exception/               # 异常目录
│   ├── QueueException.php   # 队列异常基类
│   ├── DriverException.php  # 驱动异常
│   └── TransactionException.php # 事务异常
├── Middleware/              # 中间件目录
│   ├── MiddlewareInterface.php # 中间件接口
│   ├── LogMiddleware.php    # 日志中间件
│   ├── RetryMiddleware.php  # 重试中间件
│   └── RateLimitMiddleware.php # 限流中间件
├── Context/                 # 上下文目录
│   └── Context.php          # 队列操作上下文
├── Util/                    # 工具目录
│   └── QueueUtil.php        # 队列工具类
├── AbstractQueue.php        # 抽象队列基类
├── Factory.php              # 工厂类
└── QueueInterface.php       # 队列接口

测试

# 运行测试
composer test

# 生成测试覆盖率报告
composer test-coverage