huahuibao/common/components/rabbitmq/customConsumer.php

<?php
/**
 * 客户操作消费者
 * 处理客户创建、修改等业务操作
 */

namespace common\components\rabbitmq;

use bizGhs\custom\classes\CustomClass;
use bizHd\custom\classes\CustomClass as HdCustomClass;
use common\components\noticeUtil;
use mikemadisonweb\rabbitmq\components\ConsumerInterface;
use PhpAmqpLib\Message\AMQPMessage;

class customConsumer 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}", '15280215347');
                return ConsumerInterface::MSG_REJECT;
            }
            print_r($data);
            // 根据操作类型分发处理
            $type = $data['type'] ?? null;
            $result = $this->runWithDbReconnect(function () use ($type, $data) {
                switch ($type) {
                    case 'add_custom':
                        //添加新客户
                        return CustomClass::generateCustom($data);
                    case 'pull_custom_from_other_shop':
                        //从分店拉取客户
                        return CustomClass::pullOtherShopCustom($data);
                    case 'hd_change_custom_expense_level':
                        //客户消费等级变更
                        return HdCustomClass::updateCustomExpenseLevel($data);
                    default:
                        noticeUtil::push("客户的消费者报错，不存在的类型: {$type}");
                        return false;
                }
            });
            if ($result) {
                return ConsumerInterface::MSG_ACK;
            } else {
                noticeUtil::push("客户的消费者报错：Notify message processing failed", '15280215347');
                return ConsumerInterface::MSG_REQUEUE;
            }

        } catch (\Exception $e) {
            noticeUtil::push("客户的消费者报错：" . $e->getMessage());
            return ConsumerInterface::MSG_REQUEUE;
        }
    }

}