diff --git a/Server/application/command/SyncAllFriendsCommand.php b/Server/application/command/SyncAllFriendsCommand.php new file mode 100644 index 00000000..e4b82d88 --- /dev/null +++ b/Server/application/command/SyncAllFriendsCommand.php @@ -0,0 +1,67 @@ +setName('sync:allFriends') + ->setDescription('同步所有好友(自动分页队列)') + ->addOption('jobId', null, Option::VALUE_OPTIONAL, '任务ID', date('YmdHis') . rand(1000, 9999)); + } + + protected function execute(Input $input, Output $output) + { + $output->writeln('开始同步所有好友...'); + try { + $jobId = $input->getOption('jobId'); + $queueLockKey = "queue_lock:{$this->queueName}"; + Cache::rm($queueLockKey); + if (Cache::get($queueLockKey)) { + $output->writeln("队列 {$this->queueName} 已经在运行中,跳过执行"); + Log::warning("队列 {$this->queueName} 已经在运行中,跳过执行"); + return false; + } + Cache::set($queueLockKey, $jobId, 3600); + $output->writeln("已设置队列运行锁,键名:{$queueLockKey},值:{$jobId},有效期:1小时"); + + $pageSize = 1000; + $accounts = Db::table('s2_wechat_account')->where('wechatAlive', 1)->select(); + foreach ($accounts as $account) { + $this->addToQueue($account['wechatId'], 0, $pageSize, '', $jobId, $queueLockKey); + } + + $output->writeln('同步所有好友任务已添加到队列'); + } catch (\Exception $e) { + Log::error('同步所有好友任务添加失败:' . $e->getMessage()); + $output->writeln('同步所有好友任务添加失败:' . $e->getMessage()); + return false; + } + return true; + } + + public function addToQueue($wechatId, $pageIndex, $pageSize, $preFriendId, $jobId, $queueLockKey) + { + $data = [ + 'wechatId' => $wechatId, + 'pageIndex' => $pageIndex, + 'pageSize' => $pageSize, + 'preFriendId' => $preFriendId, + 'jobId' => $jobId, + 'queueLockKey' => $queueLockKey + ]; + Queue::push(SyncAllFriendsJob::class, $data, $this->queueName); + } +} \ No newline at end of file diff --git a/Server/application/job/SyncAllFriendsJob.php b/Server/application/job/SyncAllFriendsJob.php new file mode 100644 index 00000000..1ae2c8bd --- /dev/null +++ b/Server/application/job/SyncAllFriendsJob.php @@ -0,0 +1,70 @@ +getlist([ + 'wechatAccountKeyword' => $wechatId, + 'pageIndex' => $pageIndex, + 'pageSize' => $pageSize, + 'preFriendId' => $preFriendId + ], true); + + if (is_string($result)) { + $result = json_decode($result, true); + } + + if ($result['code'] == 200) { + $friends = $result['data']['list'] ?? $result['data']; + if (is_array($friends) && count($friends) == $pageSize) { + $lastFriendId = $friends[count($friends) - 1]['id']; + // 还有下一页,重新入队 + $nextPageIndex = $pageIndex + 1; + \think\Queue::push(self::class, [ + 'wechatId' => $wechatId, + 'pageIndex' => $nextPageIndex, + 'pageSize' => $pageSize, + 'preFriendId' => $lastFriendId, + 'jobId' => $jobId, + 'queueLockKey' => $queueLockKey + ], $job->getQueue()); + Log::info("微信id: {$wechatId} 下一页任务已入队,pageIndex: {$nextPageIndex},preFriendId: {$lastFriendId}"); + } + } + + $job->delete(); + Log::info('同步微信id: ' . $wechatId . ' 第' . $pageIndex . '页任务执行成功'); + // 释放锁逻辑可在所有账号所有分页都完成后处理 + return true; + } catch (\Exception $e) { + Log::error('同步所有好友任务异常:' . $e->getMessage()); + if (!empty($data['queueLockKey'])) { + \think\facade\Cache::rm($data['queueLockKey']); + } + if ($job->attempts() > 3) { + $job->delete(); + } else { + $job->release(\think\facade\Config::get('queue.failed_delay', 10)); + } + return false; + } + } +} \ No newline at end of file