ensureDbConnection(); // 反序列化消息体 $data = unserialize($msg->body); if (!is_array($data)) { noticeUtil::push("库存的消费者报错:Invalid notify message format: {$msg->body}"); return ConsumerInterface::MSG_REJECT; } print_r($data); // 根据操作类型分发处理 $type = $data['type'] ?? null; if ($type == 'limit_buy_clear') { Yii::info('限购延迟消息开始消费: ' . json_encode($data, JSON_UNESCAPED_UNICODE), __METHOD__); Yii::getLogger()->flush(true); } switch ($type) { case 'add': $result = true; echo '持久化OK---'; break; case 'limit_buy_clear': echo 'limit_buy_clear---'; $result = $this->runWithDbReconnect(function () use ($data) { return $this->clearOrderItemLimitBuy($data); }); break; case 'birthday_gift_expire': $giftId = intval($data['giftId']); echo 'birthday_gift_expire --- giftId=' . $giftId; $result = $this->runWithDbReconnect(function () use ($giftId) { return BirthdayGiftClass::expireById($giftId); }); break; case 'group_buy_expire': $groupBuyId = intval($data['groupBuyId']); echo 'group_buy_expire --- groupBuyId=' . $groupBuyId; $result = $this->runWithDbReconnect(function () use ($groupBuyId) { return GroupBuyClass::expireById($groupBuyId); }); break; default: noticeUtil::push("库存的消费者报错,未知 type: {$type}"); $result = false; } if ($result) { return ConsumerInterface::MSG_ACK; } else { noticeUtil::push("库存的消费者报错:Stock message processing failed"); //return ConsumerInterface::MSG_REQUEUE; return ConsumerInterface::MSG_ACK; } } catch (\Exception $e) { noticeUtil::push("库存的消费者报错:" . $e->getMessage()); //return ConsumerInterface::MSG_REQUEUE; return ConsumerInterface::MSG_ACK; } } /** * 清空订单项限购值 * * @param array $data * @return bool */ private function clearOrderItemLimitBuy($data) { $productId = intval($data['productId']); if ($productId <= 0) { noticeUtil::push('取消限购的消费者报错:limit_buy_clear 缺少 productId'); return true; } $clearAt = intval($data['clearAt'] ?? 0); if ($clearAt <= 0) { noticeUtil::push('取消限购的消费者报错:limit_buy_clear 缺少 clearAt'); return true; } $ptType = $data['ptType'] ?? 'ghs'; $productClass = $ptType == 'ghs' ? ProductClass::class : hdProductClass::class; // 旧消息直接忽略,避免“先到期的旧消息”提前清空 if (!$productClass::checkLimitBuyClearMessage($productId, $clearAt)) { return true; } $recurring = intval($data['recurring'] ?? 0); if ($recurring == 1) { $result = $productClass::clearLimitBuyRecordByProductId($productId); if ($result) { $intervalDays = intval($data['intervalDays'] ?? 0); $clearHour = intval($data['clearHour'] ?? -1); if ($intervalDays <= 0 || $clearHour < 0 || $clearHour > 23) { $config = $productClass::getLimitBuyClearLoopConfigByProductId($productId); $intervalDays = intval($config['intervalDays'] ?? 0); $clearHour = intval($config['clearHour'] ?? -1); } if ($intervalDays > 0 && $clearHour >= 0 && $clearHour <= 23) { $productClass::createRecurringLimitBuyCache($productId, $intervalDays, $clearHour); } else { $productClass::clearLimitBuyClearMark($productId); noticeUtil::push('循环限购清理成功但缺少下一次调度配置: ' . json_encode([ 'ptType' => $ptType, 'productId' => $productId, 'clearAt' => $clearAt, ], JSON_UNESCAPED_UNICODE)); } } else { noticeUtil::push('限购字段清理失败: ' . json_encode([ 'ptType' => $ptType, 'productId' => $productId, 'clearAt' => $clearAt, ], JSON_UNESCAPED_UNICODE)); } return $result; } $result = $productClass::clearLimitBuyByProductId($productId); if ($result) { $productClass::clearLimitBuyClearMark($productId); } else { noticeUtil::push('限购字段清理失败: ' . json_encode([ 'ptType' => $ptType, 'productId' => $productId, 'clearAt' => $clearAt, ], JSON_UNESCAPED_UNICODE)); } return $result; } }