PHP 怎么PHP 公平队列

本文目录导读：

1. 基于优先级的 SplPriorityQueue
2. 基于时间片的轮询队列
3. 使用 Redis 实现分布式公平队列
4. 使用 Laravel 的队列系统实现公平
5. 选择建议
6. 关键设计原则

在 PHP 中实现“公平队列”，通常指的是按优先级或按资源占用比例来调度任务，确保高优先级任务不会无限等待,同时低优先级任务也能得到执行机会。

PHP 本身没有内置“公平队列”类,但可以通过以下几种方式实现：

基于优先级的 SplPriorityQueue

PHP 内置的 SplPriorityQueue 可以实现优先队列，但默认是“绝对优先级”模式（高优先级永远先执行），要让它“公平”,需要调整优先级计算逻辑。

<?php
class FairPriorityQueue extends SplPriorityQueue
{
    // 公平性：记录每个优先级最后执行的时间，动态调整优先级
    private array $lastExecutedTime = [];
    private int $basePriority;
    public function __construct(int $basePriority = 100)
    {
        $this->basePriority = $basePriority;
    }
    public function insert(mixed $value, mixed $priority): void
    {
        // 存储原始优先级和插入时间
        parent::insert([
            'value' => $value,
            'priority' => $priority,
            'insert_time' => microtime(true)
        ], $priority);
    }
    public function extract(): mixed
    {
        $data = parent::extract();
        $priority = $data['priority'];
        // 记录执行时间
        $this->lastExecutedTime[$priority] = microtime(true);
        return $data['value'];
    }
    // 公平调度：根据等待时间和优先级动态计算
    public function compare($priority1, $priority2): int
    {
        // 动态优先级 = 原始优先级 + 等待时间补偿
        $waitTime1 = microtime(true) - $this->lastExecutedTime[$priority1] ?? 0;
        $waitTime2 = microtime(true) - $this->lastExecutedTime[$priority2] ?? 0;
        // 等待时间越长，等效优先级越高
        $effectiveP1 = $priority1 + ($waitTime1 * 10); // 权重可调
        $effectiveP2 = $priority2 + ($waitTime2 * 10);
        return $effectiveP1 <=> $effectiveP2;
    }
}
// 使用示例
$queue = new FairPriorityQueue();
$queue->insert('low_task', 1);   // 低优先级
$queue->insert('high_task', 10); // 高优先级
$queue->insert('mid_task', 5);   // 中优先级
while ($queue->count() > 0) {
    echo $queue->extract() . "\n";
    // 高优先级执行后，低优先级会逐渐提升等效优先级
}

基于时间片的轮询队列

适合需要公平分配 CPU 或资源使用量的场景：

<?php
class TimeSliceFairQueue
{
    private array $queues = []; // 按优先级分组
    private array $timeSlice = []; // 每个优先级的时间片（秒）
    private int $currentPriority = 0;
    private float $sliceStartTime;
    public function __construct(array $timeSlices = [1 => 0.1, 2 => 0.05])
    {
        $this->timeSlice = $timeSlices;
        $this->sliceStartTime = microtime(true);
    }
    public function enqueue($item, int $priority = 1): void
    {
        if (!isset($this->queues[$priority])) {
            $this->queues[$priority] = new SplQueue();
        }
        $this->queues[$priority]->enqueue($item);
    }
    public function dequeue(): mixed
    {
        if (empty($this->queues)) return null;
        // 检查当前优先级是否超时
        $elapsed = microtime(true) - $this->sliceStartTime;
        if ($elapsed > ($this->timeSlice[$this->currentPriority] ?? 0.1)) {
            $this->rotatePriority();
        }
        // 从当前优先级队列获取任务
        $queue = &$this->queues[$this->currentPriority];
        if ($queue && !$queue->isEmpty()) {
            $item = $queue->dequeue();
            if ($queue->isEmpty()) {
                unset($this->queues[$this->currentPriority]);
                $this->rotatePriority();
            }
            return $item;
        }
        // 如果当前队列为空，切换到下一个
        $this->rotatePriority();
        return $this->dequeue();
    }
    private function rotatePriority(): void
    {
        $priorities = array_keys($this->queues);
        if (empty($priorities)) {
            $this->currentPriority = 0;
            return;
        }
        // 轮转到下一个优先级
        $currentIndex = array_search($this->currentPriority, $priorities);
        $nextIndex = ($currentIndex === false) ? 0 : ($currentIndex + 1) % count($priorities);
        $this->currentPriority = $priorities[$nextIndex];
        $this->sliceStartTime = microtime(true);
    }
}

使用 Redis 实现分布式公平队列

对于分布式系统，可以使用 Redis 的有序集合（Sorted Set）实现公平队列：

<?php
class RedisFairQueue
{
    private \Redis $redis;
    private string $queueKey;
    public function __construct(\Redis $redis, string $queueKey = 'fair_queue')
    {
        $this->redis = $redis;
        $this->queueKey = $queueKey;
    }
    // 添加任务，priority 越高越优先
    public function push(string $task, int $priority = 0): void
    {
        // 使用时间戳保证公平性：优先级 + 时间戳微调
        $score = ($priority * 1000000) + (PHP_INT_MAX - microtime(true) * 1000000);
        $this->redis->zAdd($this->queueKey, $score, $task);
    }
    // 取出最高优先级的任务
    public function pop(): ?string
    {
        $tasks = $this->redis->zRange($this->queueKey, 0, 0);
        if (empty($tasks)) return null;
        $task = $tasks[0];
        $this->redis->zRem($this->queueKey, $task);
        return $task;
    }
    // 公平取出（防止饥饿）
    public function fairPop(): ?string
    {
        // 获取所有任务及其分数
        $tasks = $this->redis->zRange($this->queueKey, 0, -1, true);
        if (empty($tasks)) return null;
        // 计算每个任务的等待时间（基于插入时间）
        $now = microtime(true) * 1000000;
        $bestTask = null;
        $bestScore = PHP_FLOAT_MAX;
        foreach ($tasks as $task => $score) {
            $priority = floor($score / 1000000);
            $insertTime = PHP_INT_MAX - ($score % 1000000);
            $waitTime = $now - $insertTime;
            // 公平得分 = 优先级权重 + 等待时间权重
            $fairScore = 1 / ($priority + 1) + ($waitTime / 1000000) * 0.1;
            if ($fairScore < $bestScore) {
                $bestScore = $fairScore;
                $bestTask = $task;
            }
        }
        if ($bestTask) {
            $this->redis->zRem($this->queueKey, $bestTask);
        }
        return $bestTask;
    }
}

使用 Laravel 的队列系统实现公平

如果你使用 Laravel,可以利用其队列系统的优先级和延迟功能：

// 1. 定义不同的队列名称
dispatch(new HighPriorityJob())->onQueue('high');
dispatch(new LowPriorityJob())->onQueue('low');
// 2. 配置 worker 按比例消费
// 在 supervisor 配置中设置不同权重
// [program:worker-high]
// command=php artisan queue:work --queue=high,low --sleep=3 --tries=3
// numprocs=3  # 高优先级分配更多进程
// 3. 或者使用队列的延迟平衡策略
dispatch((new Job())->delay(now()->addMinutes(5))); // 低优先级任务延迟执行

选择建议

关键设计原则

1. 防止饥饿：确保低优先级任务最终也能获得执行机会
2. 动态权重：根据等待时间调整优先级
3. 可配置：让用户能调整公平性参数（如时间片大小、权重系数）
4. 监控：记录每个优先级的等待时间和执行次数

实现公平队列的核心在于平衡不同优先级任务的执行机会，而不是严格保证 FIFO 顺序,根据你的具体业务需求选择合适的实现方式。