PHP 租户队列怎么隔离

wen PHP项目 2026-08-10 227

本文目录导读：

PHP 租户队列怎么隔离

数据库级隔离（最推荐）
Redis 队列隔离
消息队列中间件方案
公平调度与租户隔离结合
监控与治理
建议

在 PHP 中实现租户队列隔离，核心目标是让不同租户的数据互不干扰，同时保证资源分配的公平性，以下是几种常见且有效的隔离方案：

数据库级隔离（最推荐）

按租户分表/分库

class QueueService
{
    private $tenantId;
    public function __construct($tenantId)
    {
        $this->tenantId = $tenantId;
    }
    // 每个租户使用独立的队列表
    public function getQueueTable(): string
    {
        return "queue_tenant_{$this->tenantId}";
    }
    public function push(string $job, array $data)
    {
        $table = $this->getQueueTable();
        // 写入对应的租户表
        DB::table($table)->insert([
            'job' => $job,
            'data' => json_encode($data),
            'created_at' => now()
        ]);
    }
    public function pop()
    {
        $table = $this->getQueueTable();
        // 从租户表中取出任务
        return DB::table($table)
            ->where('status', 'pending')
            ->lockForUpdate()
            ->first();
    }
}

使用租户ID字段

class MultiTenantQueue
{
    public function push(int $tenantId, string $job, array $data)
    {
        DB::table('jobs')->insert([
            'tenant_id' => $tenantId,
            'job' => $job,
            'data' => json_encode($data),
            'created_at' => now()
        ]);
    }
    // 队列消费者
    public function consume(int $tenantId)
    {
        return DB::table('jobs')
            ->where('tenant_id', $tenantId)
            ->where('status', 'pending')
            ->orderBy('id')
            ->limit(1)
            ->lockForUpdate()
            ->get();
    }
}

Redis 队列隔离

使用不同的 Redis Key 前缀

class RedisQueueIsolation
{
    private $redis;
    public function __construct()
    {
        $this->redis = new Redis();
        $this->redis->connect('127.0.0.1', 6379);
    }
    // 每个租户有独立的队列
    public function getQueueKey(int $tenantId): string
    {
        return "queue:tenant:{$tenantId}";
    }
    public function pushJob(int $tenantId, $job)
    {
        $key = $this->getQueueKey($tenantId);
        return $this->redis->rPush($key, serialize($job));
    }
    public function popJob(int $tenantId)
    {
        $key = $this->getQueueKey($tenantId);
        $job = $this->redis->lPop($key);
        return $job ? unserialize($job) : null;
    }
    // 获取租户队列长度（用于监控和限流）
    public function getQueueLength(int $tenantId): int
    {
        $key = $this->getQueueKey($tenantId);
        return $this->redis->lLen($key);
    }
}

使用 Redis Hash 存储多租户队列

class HashBasedIsolation
{
    private $redis;
    private $hashKey = 'multi_tenant_queue';
    public function pushJob(int $tenantId, $jobData, $priority = 0)
    {
        $jobId = uniqid('job_', true);
        $this->redis->hSet($this->hashKey, "{$tenantId}:{$jobId}", json_encode([
            'data' => $jobData,
            'priority' => $priority,
            'tenant_id' => $tenantId,
            'created_at' => time()
        ]));
        return $jobId;
    }
    public function getTenantJobs(int $tenantId)
    {
        return $this->redis->hGetAll($this->hashKey);
    }
}

消息队列中间件方案

RabbitMQ 多vhost/Exchange

class RabbitMQIsolation
{
    private $connection;
    private $channel;
    public function __construct()
    {
        $this->connection = new AMQPConnection([
            'host' => 'localhost',
            'port' => 5672,
            'login' => 'user',
            'password' => 'pass'
        ]);
        $this->connection->connect();
        $this->channel = new AMQPChannel($this->connection);
    }
    // 每个租户一个独立队列
    public function declareTenantQueue(int $tenantId)
    {
        $queueName = "tenant_queue_{$tenantId}";
        $queue = new AMQPQueue($this->channel);
        $queue->setName($queueName);
        $queue->setFlags(AMQP_DURABLE);
        $queue->declareQueue();
        return $queue;
    }
    public function publishToTenantQueue(int $tenantId, $message)
    {
        $exchangeName = "tenant_exchange_{$tenantId}";
        $exchange = new AMQPExchange($this->channel);
        $exchange->setName($exchangeName);
        $exchange->setType(AMQP_EX_TYPE_DIRECT);
        $exchange->declareExchange();
        $exchange->publish(
            json_encode($message),
            "route_{$tenantId}",
            AMQP_NOPARAM,
            ['delivery_mode' => AMQP_DURABLE]
        );
    }
}

使用 Laravel Queue + 多连接

// config/queue.php
'connections' => [
    'tenant_1' => [
        'driver' => 'redis',
        'connection' => 'default',
        'queue' => 'tenant_1_queue',
    ],
    'tenant_2' => [
        'driver' => 'redis',
        'connection' => 'default',
        'queue' => 'tenant_2_queue',
    ],
],
// 使用时动态选择连接
class TenantQueueManager
{
    public function dispatch(int $tenantId, $job)
    {
        $connection = "tenant_{$tenantId}_connection";
        Queue::connection($connection)->push($job);
    }
}

公平调度与租户隔离结合

class FairQueueScheduler
{
    private $tenantWeights = [];
    public function __construct()
    {
        // 租户权重配置（可动态调整）
        $this->tenantWeights = [
            1 => 10,  // 租户1 权重10
            2 => 5,   // 租户2 权重5
            3 => 3,   // 租户3 权重3
        ];
    }
    public function getNextTenant()
    {
        // 简单轮询 + 权重
        $tenants = array_keys($this->tenantWeights);
        $totalWeight = array_sum($this->tenantWeights);
        $rand = mt_rand(1, $totalWeight);
        $counter = 0;
        foreach ($this->tenantWeights as $tenant => $weight) {
            $counter += $weight;
            if ($rand <= $counter) {
                return $tenant;
            }
        }
        return $tenants[0];
    }
    public function processNextJob()
    {
        $tenantId = $this->getNextTenant();
        // 从对应租户队列中取任务并处理
        return $this->processTenantJob($tenantId);
    }
}

监控与治理

class QueueMonitor
{
    public function getTenantStats(int $tenantId): array
    {
        return [
            'pending_count' => $this->getPendingCount($tenantId),
            'processing_count' => $this->getProcessingCount($tenantId),
            'failed_count' => $this->getFailedCount($tenantId),
            'avg_wait_time' => $this->getAverageWaitTime($tenantId),
        ];
    }
    public function alertForAbnormalUsage(int $tenantId, float $threshold = 0.8)
    {
        $usage = $this->getTenantResourceUsage($tenantId);
        $maxAllowed = $this->getTenantLimit($tenantId);
        if ($usage > ($maxAllowed * $threshold)) {
            // 发送告警或限制租户
            $this->rateLimitTenant($tenantId);
        }
    }
}

建议

根据业务规模选择：
- 小规模（<100租户）：数据库单表加索引，使用租户ID字段
- 中规模（100-1000租户）：Redis多队列或数据库分表
- 大规模（>1000租户）：RabbitMQ/Kafka等专业MQ，按租户分区
考虑隔离级别：
- 数据隔离：不同租户队列数据完全隔离
- 性能隔离：限制单个租户的资源使用
- 故障隔离：一个租户的异常不影响其他租户
动态扩容：
- 支持租户队列的动态创建和销毁
- 支持根据租户负载动态调整优先级