huahuibao/common/components/rabbitmq/stockConsumer.php

<?php
/**
 * 库存管理消费者
 * 处理花材库存补充、扣减等库存操作
 */

namespace common\components\rabbitmq;

use bizGhs\product\classes\ProductClass;
use bizHd\birthday\classes\BirthdayGiftClass;
use bizHd\groupBuy\classes\GroupBuyClass;
use bizHd\product\classes\ProductClass as hdProductClass;
use common\components\noticeUtil;
use mikemadisonweb\rabbitmq\components\ConsumerInterface;
use PhpAmqpLib\Message\AMQPMessage;
use Yii;

class stockConsumer extends baseConsumer
{
    /**
     * 执行消费者逻辑
     * 
     * @param AMQPMessage $msg 消息对象
     * @return string 消息处理结果
     * 
     * ConsumerInterface::MSG_ACK - 确认消息（标记为已处理）并从队列中删除
     * ConsumerInterface::MSG_REJECT - 拒绝并从队列中删除消息
     * ConsumerInterface::MSG_REQUEUE - 拒绝并重新入队消息
     */
    public function execute(AMQPMessage $msg)
    {
        try {
            $this->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;
    }

}