huahuibao/common/components/rabbitmq/notifyConsumer.php

<?php
/**
 * 消息通知消费者
 * 处理各种消息通知：新订单、新客户、充值、销账、配送状态变更等
 */

namespace common\components\rabbitmq;

use common\components\noticeUtil;
use mikemadisonweb\rabbitmq\components\ConsumerInterface;
use PhpAmqpLib\Message\AMQPMessage;
use Yii;

class notifyConsumer implements ConsumerInterface
{
    /**
     * 执行消费者逻辑
     *
     * @param AMQPMessage $msg 消息对象
     * @return string 消息处理结果
     *
     * ConsumerInterface::MSG_ACK - 确认消息（标记为已处理）并从队列中删除
     * ConsumerInterface::MSG_REJECT - 拒绝并从队列中删除消息
     * ConsumerInterface::MSG_REQUEUE - 拒绝并重新入队消息
     */
    public function execute(AMQPMessage $msg)
    {
        try {
            // 反序列化消息体
            $data = unserialize($msg->body);
            if (!is_array($data)) {
                noticeUtil::push("通知的消费者报错：Invalid notify message format: {$msg->body}", '15280215347');
                return ConsumerInterface::MSG_REJECT;
            }
            // 根据通知类型分发处理
            $type = $data['type'] ?? null;
            switch ($type) {
                case 'ghs_new_order':
                    //供货商的新订单通知
                    $result = $this->ghsNewOrder($data);
                    break;
                case 'hd_new_order':
                    //花店的新订单通知
                    $result = $this->hdNewOrder($data);
                    break;
                case 'hd_new_cg_order':
                    //花店的新采购单通知
                    $result = $this->hdNewCgOrder($data);
                    break;
                default:
                    noticeUtil::push("通知的消费者提示：Unknown notify type: {$type}", '15280215347');
                    $result = false;
            }
            if ($result) {
                return ConsumerInterface::MSG_ACK;
            } else {
                noticeUtil::push("通知的消费者提示：Notify message processing failed", '15280215347');
                return ConsumerInterface::MSG_REQUEUE;
            }
        } catch (\Exception $e) {
            noticeUtil::push("Notify consumer exception: " . $e->getMessage(), '15280215347');
            return ConsumerInterface::MSG_REQUEUE;
        }
    }

    private function ghsNewOrder($data)
    {
        $msg = $data['msg'] ?? '';
        noticeUtil::push($msg, '15280215347');
        return true;
    }

    private function hdNewOrder($data)
    {
        $msg = $data['msg'] ?? '';
        noticeUtil::push($msg, '15280215347');
        return true;
    }

    private function hdNewCgOrder($data)
    {
        $msg = $data['msg'] ?? '';
        noticeUtil::push($msg, '15280215347');
        return true;
    }

}