From 336fa08a18c3d29f292cc4a2c9e779e4a0b73080 Mon Sep 17 00:00:00 2001 From: wangjinlei <751475802@qq.com> Date: Fri, 8 May 2026 15:38:15 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E6=94=B9=E8=87=AA=E5=8A=A8=E6=8E=A8?= =?UTF-8?q?=E5=B9=BF=E7=9A=84=E7=9B=B8=E5=85=B3=E4=BB=BB=E5=8A=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- application/api/controller/EmailClient.php | 77 ++++++++++++++++--- application/api/job/FetchExperts.php | 18 ----- application/api/job/FillExpertCountry.php | 16 ---- application/api/job/PromotionPrepare.php | 19 ----- application/api/job/PromotionPrepareEmail.php | 31 +------- application/api/job/PromotionPrepareTask.php | 23 ------ application/api/job/PromotionSend.php | 22 ------ application/api/job/myQueue.php | 31 ++++++++ application/common/PromotionService.php | 30 +++++--- 9 files changed, 116 insertions(+), 151 deletions(-) create mode 100644 application/api/job/myQueue.php diff --git a/application/api/controller/EmailClient.php b/application/api/controller/EmailClient.php index 12a402f..760482e 100644 --- a/application/api/controller/EmailClient.php +++ b/application/api/controller/EmailClient.php @@ -1761,20 +1761,9 @@ class EmailClient extends Base $service = new PromotionService(); $taskId = intval($data['id']); - // 调用前快照：用于解释"为什么没入队" - $task = \think\Db::name('promotion_task')->where('task_id', $taskId)->find(); - $pending = \think\Db::name('promotion_email_log') - ->where('task_id', $taskId) - ->where('state', 0) - ->where('prepared_at', 0) - ->count(); - $result = $service->dispatchPrepareEmails($taskId); return jsonSuccess([ - 'task_id' => $taskId, - 'task_state' => $task ? intval($task['state']) : null, // 0 才能 dispatch；5 已准备完 - 'pending_before' => intval($pending), // 调用前还能入队的 log 数 'dispatch_result' => $result, // ['dispatched' => N, ...] ]); } @@ -1793,6 +1782,72 @@ class EmailClient extends Base return jsonSuccess($result); } + public function mytestqqq(){ + $data = $this->request->post(); + $rule = new Validate([ + "id"=>"require" + ]); + if(!$rule->check($data)){ + return jsonError($rule->getError()); + } + $service = new PromotionService(); + $service->enqueuePrepareEmail(intval($data['id'])); + return jsonSuccess(); + } + + /** + * 队列写入测试：推一条 myTestQueue，并返回 push 返回值 + 与 queueDebug 相同 Redis 下的 LLEN（便于对照 redis-cli）。 + */ + public function mytestQueue() + { + $queueName = 'myTestQueue'; + $jobClass = 'app\\api\\job\\myQueue@fire'; + $data = ['testData' => 1]; + + $cfg = Config::get('queue'); + $connector = isset($cfg['connector']) ? (string)$cfg['connector'] : ''; + if (stripos($connector, 'sync') !== false) { + return jsonError('queue connector is Sync — will not write Redis. Check application/extra/queue.php'); + } + if (!extension_loaded('redis')) { + return jsonError('PHP redis extension not loaded — Queue Redis connector cannot run.'); + } + + try { + $jobId = Queue::push($jobClass, $data, $queueName); + } catch (\Throwable $e) { + return jsonError('Queue::push failed: ' . $e->getMessage()); + } + + // 独立连一次 Redis，确认与 PHP-FPM 使用的配置一致且 LIST 长度（避免「以为写了其实没有」） + $verify = ['ready_len' => null, 'error' => null]; + try { + $redis = new \Redis(); + $connectMethod = !empty($cfg['persistent']) ? 'pconnect' : 'connect'; + $redis->$connectMethod($cfg['host'] ?? '127.0.0.1', $cfg['port'] ?? 6379); + if (!empty($cfg['password'])) { + $redis->auth($cfg['password']); + } + $redis->select($cfg['select'] ?? 0); + $readyKey = 'queues:' . $queueName; + $verify['ready_len'] = (int)$redis->lLen($readyKey); + $verify['exists'] = (bool)$redis->exists($readyKey); + $verify['ping'] = (string)$redis->ping(); + $verify['checked_db'] = (int)($cfg['select'] ?? 0); + $verify['checked_host'] = (string)($cfg['host'] ?? '127.0.0.1'); + $verify['checked_port'] = (int)($cfg['port'] ?? 6379); + } catch (\Throwable $e) { + $verify['error'] = $e->getMessage(); + } + + return jsonSuccess([ + 'job_id' => $jobId, + 'queue' => $queueName, + 'connector' => $connector, + 'after_push' => $verify, + ]); + } + /** * 队列调试：查看 Redis 里队列长度（不依赖 redis-cli）。 * diff --git a/application/api/job/FetchExperts.php b/application/api/job/FetchExperts.php index f909401..83be73f 100644 --- a/application/api/job/FetchExperts.php +++ b/application/api/job/FetchExperts.php @@ -4,7 +4,6 @@ namespace app\api\job; use think\queue\Job; use app\common\ExpertFinderService; -use app\common\QueueJob; /** * 专家抓取队列任务。 @@ -16,25 +15,16 @@ use app\common\QueueJob; */ class FetchExperts { - private $oQueueJob; - - public function __construct() - { - $this->oQueueJob = new QueueJob(); - } public function fire(Job $job, $data) { - $this->oQueueJob->init($job); $field = isset($data['field']) ? (string)$data['field'] : ''; if ($field === '') { - $this->oQueueJob->log("FetchExperts 无效的 field，删除任务"); $job->delete(); return; } - try { $service = new ExpertFinderService(); $service->doFetchForField( $field, @@ -42,14 +32,6 @@ class FetchExperts isset($data['per_page']) ? intval($data['per_page']) : 100, isset($data['min_year']) ? $data['min_year'] : null ); - $this->oQueueJob->log("FetchExperts 完成 | field={$field}"); $job->delete(); - } catch (\Exception $e) { - $this->oQueueJob->handleException($e, $job, "field={$field}"); - } catch (\Throwable $e) { - $this->oQueueJob->handleException($e, $job, "field={$field}"); - } finally { - $this->oQueueJob->finnal(); - } } } diff --git a/application/api/job/FillExpertCountry.php b/application/api/job/FillExpertCountry.php index 993485f..c58ff12 100644 --- a/application/api/job/FillExpertCountry.php +++ b/application/api/job/FillExpertCountry.php @@ -4,7 +4,6 @@ namespace app\api\job; use think\queue\Job; use app\common\ExpertFinderService; -use app\common\QueueJob; /** * 队列任务：用本地大模型从 affiliation 推断国家，写入 expert.country_id / country。 @@ -17,16 +16,9 @@ use app\common\QueueJob; */ class FillExpertCountry { - private $oQueueJob; - - public function __construct() - { - $this->oQueueJob = new QueueJob(); - } public function fire(Job $job, $data) { - $this->oQueueJob->init($job); $expertId = intval(isset($data['expert_id']) ? $data['expert_id'] : 0); $affiliation = isset($data['affiliation']) ? trim((string)$data['affiliation']) : ''; @@ -35,7 +27,6 @@ class FillExpertCountry $service = new ExpertFinderService(); - try { if ($expertId && $affiliation !== '') { $service->fillExpertCountry($expertId, $affiliation, $chatUrl); } @@ -43,12 +34,5 @@ class FillExpertCountry // 链式：处理完当前专家立刻拉下一个进来 $service->enqueueNextCountryFill(1, $queue, $chatUrl); - } catch (\Exception $e) { - $this->oQueueJob->handleException($e, $job, "expert_id={$expertId} queue={$queue}"); - } catch (\Throwable $e) { - $this->oQueueJob->handleException($e, $job, "expert_id={$expertId} queue={$queue}"); - } finally { - $this->oQueueJob->finnal(); - } } } diff --git a/application/api/job/PromotionPrepare.php b/application/api/job/PromotionPrepare.php index 3fa05ea..4ea0fd4 100644 --- a/application/api/job/PromotionPrepare.php +++ b/application/api/job/PromotionPrepare.php @@ -4,7 +4,6 @@ namespace app\api\job; use think\queue\Job; use app\common\PromotionService; -use app\common\QueueJob; /** * 【已废弃 / 兼容保留】 @@ -16,35 +15,17 @@ use app\common\QueueJob; */ class PromotionPrepare { - private $oQueueJob; - - public function __construct() - { - $this->oQueueJob = new QueueJob(); - } public function fire(Job $job, $data) { $this->oQueueJob->init($job); - $taskId = isset($data['task_id']) ? intval($data['task_id']) : 0; if ($taskId <= 0) { - $this->oQueueJob->log("PromotionPrepare[deprecated] 无效的 task_id，删除任务"); $job->delete(); return; } - - try { $service = new PromotionService(); $service->enqueuePrepareTask($taskId); - $this->oQueueJob->log("PromotionPrepare[deprecated] forwarded task_id={$taskId} -> PromotionPrepareTask"); $job->delete(); - } catch (\Exception $e) { - $this->oQueueJob->handleException($e, $job, "[deprecated] task_id={$taskId}"); - } catch (\Throwable $e) { - $this->oQueueJob->handleException($e, $job, "[deprecated] task_id={$taskId}"); - } finally { - $this->oQueueJob->finnal(); - } } } diff --git a/application/api/job/PromotionPrepareEmail.php b/application/api/job/PromotionPrepareEmail.php index 437ae3c..5f58670 100644 --- a/application/api/job/PromotionPrepareEmail.php +++ b/application/api/job/PromotionPrepareEmail.php @@ -4,7 +4,6 @@ namespace app\api\job; use think\queue\Job; use app\common\PromotionService; -use app\common\QueueJob; /** * 队列任务：单封邮件 prepare（调用 LLM 生成个性化描述 + 渲染模板 + 写入 log）。 @@ -18,41 +17,13 @@ use app\common\QueueJob; */ class PromotionPrepareEmail { -// private $oQueueJob; - -// public function __construct() -// { -// $this->oQueueJob = new QueueJob(); -// } public function fire(Job $job, $data) { -// $this->oQueueJob->init($job); -// $logId = isset($data['log_id']) ? intval($data['log_id']) : 0; -// if ($logId <= 0) { -// $this->oQueueJob->log("PromotionPrepareEmail 无效的 log_id，删除任务"); -// $job->delete(); -// return; -// } -// -// try { $service = new PromotionService(); $service->log("id:".$logId); -// $result = $service->prepareSingleEmail($logId); -// -// $code = isset($result['code']) ? $result['code'] : ''; -// $msg = isset($result['msg']) ? $result['msg'] : ''; -// $llm = isset($result['llm_status']) ? $result['llm_status'] : ''; -// $this->oQueueJob->log("PromotionPrepareEmail 完成 | log_id={$logId} code={$code} llm_status={$llm} msg={$msg}"); - + $result = $service->prepareSingleEmail($logId); $job->delete(); -// } catch (\Exception $e) { -// $this->oQueueJob->handleException($e, $job, "log_id={$logId}"); -// } catch (\Throwable $e) { -// $this->oQueueJob->handleException($e, $job, "log_id={$logId}"); -// } finally { -// $this->oQueueJob->finnal(); -// } } } diff --git a/application/api/job/PromotionPrepareTask.php b/application/api/job/PromotionPrepareTask.php index 65d9d6d..6440730 100644 --- a/application/api/job/PromotionPrepareTask.php +++ b/application/api/job/PromotionPrepareTask.php @@ -17,40 +17,17 @@ use app\common\QueueJob; */ class PromotionPrepareTask { - private $oQueueJob; - - public function __construct() - { - $this->oQueueJob = new QueueJob(); - } public function fire(Job $job, $data) { - $this->oQueueJob->init($job); - $taskId = isset($data['task_id']) ? intval($data['task_id']) : 0; if ($taskId <= 0) { - $this->oQueueJob->log("PromotionPrepareTask 无效的 task_id，删除任务"); $job->delete(); return; } - - try { $service = new PromotionService(); $result = $service->dispatchPrepareEmails($taskId); - $dispatched = isset($result['dispatched']) ? $result['dispatched'] : 0; - $alreadyDone = isset($result['already_done']) ? $result['already_done'] : 0; - $err = isset($result['error']) ? $result['error'] : ''; - $this->oQueueJob->log("PromotionPrepareTask 完成 | task_id={$taskId} dispatched={$dispatched} already_done={$alreadyDone} error={$err}"); - $job->delete(); - } catch (\Exception $e) { - $this->oQueueJob->handleException($e, $job, "task_id={$taskId}"); - } catch (\Throwable $e) { - $this->oQueueJob->handleException($e, $job, "task_id={$taskId}"); - } finally { - $this->oQueueJob->finnal(); - } } } diff --git a/application/api/job/PromotionSend.php b/application/api/job/PromotionSend.php index 898cb21..778a6e1 100644 --- a/application/api/job/PromotionSend.php +++ b/application/api/job/PromotionSend.php @@ -4,7 +4,6 @@ namespace app\api\job; use think\queue\Job; use app\common\PromotionService; -use app\common\QueueJob; /** * 队列任务：发送 task 下"已 prepare"的邮件，按 min/max_interval 控制节奏。 @@ -16,39 +15,18 @@ use app\common\QueueJob; */ class PromotionSend { - private $oQueueJob; - - public function __construct() - { - $this->oQueueJob = new QueueJob(); - } public function fire(Job $job, $data) { - $this->oQueueJob->init($job); $taskId = isset($data['task_id']) ? intval($data['task_id']) : 0; if ($taskId <= 0) { - $this->oQueueJob->log("PromotionSend 无效的 task_id，删除任务"); $job->delete(); return; } - - try { $service = new PromotionService(); $result = $service->processNextEmail($taskId); - $done = !empty($result['done']) ? 1 : 0; - $reason = isset($result['reason']) ? $result['reason'] : ''; - $this->oQueueJob->log("PromotionSend 完成 | task_id={$taskId} done={$done} reason={$reason}"); - $job->delete(); - } catch (\Exception $e) { - $this->oQueueJob->handleException($e, $job, "task_id={$taskId}"); - } catch (\Throwable $e) { - $this->oQueueJob->handleException($e, $job, "task_id={$taskId}"); - } finally { - $this->oQueueJob->finnal(); - } } } diff --git a/application/api/job/myQueue.php b/application/api/job/myQueue.php new file mode 100644 index 0000000..fe8d4a5 --- /dev/null +++ b/application/api/job/myQueue.php @@ -0,0 +1,31 @@ +delete(); + + } catch (\Throwable $e) { + // 如果你不想重试，直接删除 + $job->delete(); + } + } + + // 可选：任务失败时执行 + public function failed($data) + { + // 失败后不做处理，直接跳过 + } + +} \ No newline at end of file diff --git a/application/common/PromotionService.php b/application/common/PromotionService.php index 9f54ca4..d958bed 100644 --- a/application/common/PromotionService.php +++ b/application/common/PromotionService.php @@ -319,19 +319,27 @@ class PromotionService $this->log("dispatchPrepareEmails task_id={$taskId} no_logs -> state=5"); return ['dispatched' => 0, 'already_done' => true, 'error' => null]; } - -// return $logIds; - - foreach ($logIds as $logId) { - echo $logId."----"; $this->enqueuePrepareEmail(intval($logId)); } - $this->log("dispatchPrepareEmails task_id={$taskId} dispatched=" . count($logIds)); return ['dispatched' => count($logIds), 'already_done' => false, 'error' => null]; } + public function prepareSingleEmailTest($logId){ + $log = Db::name('promotion_email_log')->where('log_id', $logId)->find(); +// $task = Db::name('promotion_task')->where('task_id', $log['task_id'])->find(); +// if (!$task) { +// Db::name('promotion_email_log')->where('log_id', $logId)->update([ +// 'state' => 2, +// 'error_msg' => 'Task not found', +// 'send_time' => time(), +// ]); +// return ['code' => 1, 'msg' => 'task_not_found', 'llm_status' => 0]; +// } + return jsonSuccess($log); + } + /** * 对单封邮件执行准备：拉取 expert / journal，调 LLM 生成描述，渲染模板，写回 log。 * @@ -768,16 +776,14 @@ class PromotionService * 队列名：promotion_email * 启动 worker：php think queue:listen --queue promotion_email */ - public function enqueuePrepareEmail($logId, $delay = 0) + public function enqueuePrepareEmail($logId) { $jobClass = 'app\api\job\PromotionPrepareEmail@fire'; $data = ['log_id' => intval($logId)]; - if ($delay > 0) { - Queue::later($delay, $jobClass, $data, 'PromotionPrepareEmail'); - } else { - Queue::push($jobClass, $data, 'PromotionPrepareEmail'); - } + + $res =Queue::push($jobClass, $data, 'PromotionPrepareEmail'); + } /**