84 lines
2.8 KiB
PHP
84 lines
2.8 KiB
PHP
<?php
|
||
|
||
declare(strict_types=1);
|
||
|
||
namespace app\process;
|
||
|
||
use app\common\service\GameHotDataRedis;
|
||
use app\common\service\GameHotDataWriteQueue;
|
||
use support\Redis;
|
||
use Throwable;
|
||
use Workerman\Timer;
|
||
|
||
/**
|
||
* 消费热点缓存回源队列（RPOP 批量），与 HTTP 进程分离以削峰。
|
||
*/
|
||
class GameHotDataQueueConsumer
|
||
{
|
||
public function onWorkerStart(): void
|
||
{
|
||
$interval = config('game_hot_cache.queue_consumer_tick_seconds', 0.1);
|
||
if (!is_float($interval) && !is_int($interval)) {
|
||
$interval = 0.1;
|
||
}
|
||
$interval = (float) $interval;
|
||
if ($interval < 0.02) {
|
||
$interval = 0.02;
|
||
}
|
||
|
||
Timer::add($interval, static function (): void {
|
||
if (!GameHotDataWriteQueue::enabled()) {
|
||
return;
|
||
}
|
||
$key = GameHotDataWriteQueue::queueListKey();
|
||
$batch = config('game_hot_cache.queue_consumer_batch', 80);
|
||
$n = filter_var($batch, FILTER_VALIDATE_INT);
|
||
if ($n === false || $n < 1) {
|
||
$n = 80;
|
||
}
|
||
for ($i = 0; $i < $n; $i++) {
|
||
try {
|
||
$raw = Redis::rPop($key);
|
||
} catch (Throwable) {
|
||
break;
|
||
}
|
||
if ($raw === false || $raw === null || $raw === '') {
|
||
break;
|
||
}
|
||
if (!is_string($raw)) {
|
||
continue;
|
||
}
|
||
$job = json_decode($raw, true);
|
||
if (!is_array($job)) {
|
||
continue;
|
||
}
|
||
$op = $job['op'] ?? '';
|
||
if (!is_string($op) || $op === '') {
|
||
continue;
|
||
}
|
||
try {
|
||
if ($op === GameHotDataWriteQueue::OP_USER_REFRESH) {
|
||
$uid = filter_var($job['id'] ?? null, FILTER_VALIDATE_INT);
|
||
if ($uid !== false && $uid > 0) {
|
||
GameHotDataRedis::userReplaceCacheFromDb($uid);
|
||
}
|
||
continue;
|
||
}
|
||
if ($op === GameHotDataWriteQueue::OP_GC_REFRESH) {
|
||
$k = $job['key'] ?? '';
|
||
if (is_string($k) && $k !== '') {
|
||
GameHotDataRedis::gameConfigReplaceFromDb($k);
|
||
}
|
||
continue;
|
||
}
|
||
if ($op === GameHotDataWriteQueue::OP_GR_SYNC) {
|
||
$rid = filter_var($job['id'] ?? null, FILTER_VALIDATE_INT);
|
||
GameHotDataRedis::gameRecordSyncCachesAfterDbWrite($rid === false ? null : $rid);
|
||
}
|
||
} catch (Throwable) {
|
||
}
|
||
}
|
||
});
|
||
}
|
||
}
|