本文目录导读：
在 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，按租户分区
-
考虑隔离级别：
- 数据隔离：不同租户队列数据完全隔离
- 性能隔离：限制单个租户的资源使用
- 故障隔离：一个租户的异常不影响其他租户
-
动态扩容：
- 支持租户队列的动态创建和销毁
- 支持根据租户负载动态调整优先级
选择方案时,需要权衡实现复杂度、维护成本和业务需求，通常推荐使用Redis方案，因为它简单高效，且支持大多数隔离需求。
