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(); $optType = (string)($data['optType'] ?? ''); // 加油:一单可多次加油,每笔单独通知。消息里带 consume_id 时推这一笔的占用金额和消费单号; // 老消息没有 consume_id(上线前入队的)则沿用旧口径推订单金额 $consume = null; if ($optType == 'add_oil' && !empty($data['consume_id'])) { $consume = Db::name('qyt_consume')->where('id', (int)$data['consume_id'])->find(); } if ($optType == 'refund') { $amount = $order['refund_amount']; } elseif ($consume) { $amount = number_format((int)$consume['consume_amount'] / 100, 2, '.', ''); } else { $amount = $order['total_price']; } $data = [ 'oilCardNo' => $order['oilCardNo'], // 分油:订单金额;加油:本次加油占用金额;退款:退款金额 'amount' => $amount, // 订单实付金额(即分油金额),所有场景都带,方便对接方拿 amount 对照参考 'orderAmount' => number_format((float)$order['total_price'], 2, '.', ''), 'shareNo' => $consume && $consume['order_code'] !== '' ? $consume['order_code'] : $order['shareNo'], 'type' => $optType == 'share_oil' ? 'add' : 'sub', 'optType' => $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(), ]); }elseif($data['optType'] == 'add_oil'){ // 退款通知成功不记 use_callback,避免把「加油已回调」标错 $order->save([ 'use_callback' => 1, 'use_callback_time' => time(), ]); // 逐笔记录这一笔加油已通知 if ($consume) { Db::name('qyt_consume')->where('id', $consume['id'])->update(['notify_status' => 1, 'notify_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; } } }