setName('notify:queue') ->setDescription('Process messages from the Redis queue'); } protected function execute(Input $input, Output $output) { $this->queueService = new QueueService(); while (true) { // 先迁移到期的延迟任务 $this->queueService->migrateDelayedToReady('notify_queue'); $message = $this->queueService->pop('notify_queue'); if ($message) { // 处理消息 $data = json_decode($message, true); // 如果没有重试计数,默认 0 $retryCount = isset($data['retry_count']) ? intval($data['retry_count']) : 0; $orderId = $data['order_id'] ?? null; if (!$orderId) { Log::error("队列消息缺少 order_id: " . $message); continue; } // 调用回调接口 $success = $this->notifyPartner($data); if ($success) { Log::info("订单 {$orderId} 回调成功"); } else { Log::warning("订单 {$orderId} 第 {$retryCount} 次回调失败"); if ($retryCount < 3) { $retryCount++; $delay = $this->getDelayTime($retryCount); // 重新放回延迟队列 $data['retry_count'] = $retryCount; $this->queueService->laterPush('notify_queue', json_encode($data, 320), $delay); Log::info("订单 {$orderId} 已重新入队,延迟 {$delay} 秒后再通知"); } else { Log::error("订单 {$orderId} 回调失败超过最大次数,放弃重试"); } } } else { // 暂停一段时间再检查队列 sleep(1); } } } /** * 模拟发送回调 * @param array $data * @return bool */ protected function notifyPartner($data) { $order = OrderModel::detail($data['order_id']); if(!$order){ return false; } $partners = partnersModel::detail($order['partners_id']); if(!$partners){ return false; } $partnersOil = new PartnersOil(); $data = [ 'oilCardNo' => $order['oilCardNo'], 'amount' => $order['total_price'], 'shareNo' => $order['shareNo'], 'type' => $data['optType'] == 'share_oil' ? 'add' : 'sub', 'optType' => $data['optType'], ]; $header['method'] = 'syb.notify.oilCard.change'; $res = $partnersOil->sendRequest($partners, $partners['callback'], $data, $header); $code = $res['code'] ?? ''; if($res && $code == '00'){ if($data['optType'] == 'share_oil') { $order->save([ 'oil_callback' => 1, 'oil_callback_time' => time(), ]); }else{ $order->save([ 'use_callback' => 1, 'use_callback_time' => time(), ]); } return true; } return false; } /** * 延迟时间策略 */ private function getDelayTime($retryCount) { switch ($retryCount) { case 1: return 60; // 1分钟 case 2: return 300; // 5分钟 case 3: return 1800; // 30分钟 default: return 0; } } }