diff --git a/app/common/utils/RedisUidRotator.php b/app/common/utils/RedisUidRotator.php new file mode 100644 index 0000000..44d4265 --- /dev/null +++ b/app/common/utils/RedisUidRotator.php @@ -0,0 +1,161 @@ +key = "uid_rotator:{$rotatorName}:index"; + $this->setKey = "uid_rotator:{$rotatorName}:uids"; + $this->lockKey = "uid_rotator:{$rotatorName}:lock"; + $this->expireSeconds = $expireSeconds; + } + + public function setUids(array $uids, bool $resetIndex = true): bool + { + // 删除旧数据 + Redis::del($this->setKey); + + if (!empty($uids)) { + // 添加新UID集合 + foreach ($uids as $uid) { + Redis::sAdd($this->setKey, $uid); + } + + // 重置索引 + if ($resetIndex) { + Redis::set($this->key, 0); + } + } + + $this->touchExpire(); + return true; + } + + public function next(): ?int + { + $lockAcquired = false; + try { + // 获取锁(改用原生命令) + $lockAcquired = $this->acquireLock(5); + if (!$lockAcquired) { + throw new \RuntimeException("获取锁失败"); + } + + // 获取当前索引 + $currentIndex = (int)Redis::get($this->key); + + // 获取所有UID + $uids = Redis::sMembers($this->setKey); + if (empty($uids)) { + return null; + } + + // 计算位置 + $uid = $uids[$currentIndex % count($uids)]; + $nextIndex = $currentIndex + 1; + + // 更新索引 + Redis::set($this->key, $nextIndex); + $this->touchExpire(); + + return is_numeric($uid) ? (int)$uid : $uid; + } catch (\Throwable $e) { + Log::info("UID分配错误: " . $e->getMessage()); + return null; + } finally { + if ($lockAcquired) { + Redis::del($this->lockKey); + } + } + } + + private function acquireLock(int $timeout = 5): bool + { + $start = time(); + $lockTimeout = $timeout + 1; // 锁过期时间比等待时间长 + + while (time() - $start < $timeout) { + // 使用SETNX实现锁(分两步) + if (Redis::setNx($this->lockKey, 1)) { + // 设置过期时间 + Redis::expire($this->lockKey, $lockTimeout); + return true; + } + + // 检查锁是否已过期但未被删除(处理死锁) + if (Redis::ttl($this->lockKey) < 0) { + Redis::del($this->lockKey); + } + + usleep(100000); // 100ms重试间隔 + } + + return false; + } + + private function touchExpire(): void + { + Redis::expire($this->key, $this->expireSeconds); + Redis::expire($this->setKey, $this->expireSeconds); + } + + public function getPosition(): int + { + return (int)Redis::get($this->key); + } + + public function resetIndex(int $position = 0): bool + { + return (bool)Redis::set($this->key, max(0, $position)); + } + + public function getUids(): array + { + return Redis::sMembers($this->setKey) ?: []; + } + + public function addUids($uids): int + { + $added = 0; + $uids = is_array($uids) ? $uids : [$uids]; + + foreach ($uids as $uid) { + $added += Redis::sAdd($this->setKey, $uid); + } + + $this->touchExpire(); + return $added; + } + + public function removeUids($uids): int + { + $removed = 0; + $uids = is_array($uids) ? $uids : [$uids]; + + foreach ($uids as $uid) { + $removed += Redis::sRem($this->setKey, $uid); + } + + return $removed; + } + + public function count(): int + { + return (int)Redis::sCard($this->setKey); + } + + public function destroy(): void + { + Redis::del([$this->key, $this->setKey, $this->lockKey]); + } +} \ No newline at end of file diff --git a/app/common/utils/email/Emails.php b/app/common/utils/email/Emails.php index 8a3db68..0bb8376 100644 --- a/app/common/utils/email/Emails.php +++ b/app/common/utils/email/Emails.php @@ -435,10 +435,11 @@ class Emails $channelInfo = InquiryChannel::where('id', $channel)->first(); // 初始化 UID 轮询器 - \app\common\utils\UidRotator::initialize([64, 63, 83, 93, 108, 118, 131, 132, 145, 150, 143, 156, 157, 158]); + $uidRotator = new \app\common\utils\RedisUidRotator('email_parsed_result_20250804'); + $uidRotator->setUids([64, 63, 83, 93, 108, 118, 131, 132, 145, 150, 143, 156, 157, 158]); EmailParsedResult::where('status', 0) - ->chunkById($batchSize, function ($records) use ($delaySeconds, $throttleAfter, $throttleSeconds, $enterUser, $channelInfo) { + ->chunkById($batchSize, function ($records) use ($delaySeconds, $throttleAfter, $throttleSeconds, $enterUser, $channelInfo, $uidRotator) { $count = 0; foreach ($records as $record) { @@ -448,7 +449,7 @@ class Emails $countryInfo = Country::where('cn_addr', $countryName)->first(); // 获取下一个分配的UID - $assignedUid = \app\common\utils\UidRotator::next(); + $assignedUid = $uidRotator->next(); // 获取分配的用户信息 $recipientInfo = User::where('id', $assignedUid)->first();