Merge branch 'feature/mzl_dev'
# Conflicts: # app/common/utils/email/Emails.php
This commit is contained in:
@@ -0,0 +1,161 @@
|
|||||||
|
<?php
|
||||||
|
|
||||||
|
namespace app\common\utils;
|
||||||
|
|
||||||
|
use support\Log;
|
||||||
|
use support\Redis;
|
||||||
|
|
||||||
|
class RedisUidRotator
|
||||||
|
{
|
||||||
|
private string $key;
|
||||||
|
private string $setKey;
|
||||||
|
private string $lockKey;
|
||||||
|
private int $expireSeconds;
|
||||||
|
|
||||||
|
public function __construct(string $rotatorName, int $expireSeconds = 604800)
|
||||||
|
{
|
||||||
|
$this->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]);
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -435,10 +435,11 @@ class Emails
|
|||||||
$channelInfo = InquiryChannel::where('id', $channel)->first();
|
$channelInfo = InquiryChannel::where('id', $channel)->first();
|
||||||
|
|
||||||
// 初始化 UID 轮询器
|
// 初始化 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)
|
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;
|
$count = 0;
|
||||||
|
|
||||||
foreach ($records as $record) {
|
foreach ($records as $record) {
|
||||||
@@ -448,7 +449,7 @@ class Emails
|
|||||||
$countryInfo = Country::where('cn_addr', $countryName)->first();
|
$countryInfo = Country::where('cn_addr', $countryName)->first();
|
||||||
|
|
||||||
// 获取下一个分配的UID
|
// 获取下一个分配的UID
|
||||||
$assignedUid = \app\common\utils\UidRotator::next();
|
$assignedUid = $uidRotator->next();
|
||||||
|
|
||||||
// 获取分配的用户信息
|
// 获取分配的用户信息
|
||||||
$recipientInfo = User::where('id', $assignedUid)->first();
|
$recipientInfo = User::where('id', $assignedUid)->first();
|
||||||
|
|||||||
Reference in New Issue
Block a user