huahuibao/common/components/rabbitmq/customConsumer.php

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

namespace common\components\rabbitmq;

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

class customConsumer 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)) {
                Yii::error("Invalid custom message format: {$msg->body}", 'rabbitmq.custom');
                return ConsumerInterface::MSG_REJECT;
            }
            
            Yii::info("Processing custom message: " . json_encode($data), 'rabbitmq.custom');
            
            // 根据操作类型分发处理
            $action = $data['action'] ?? null;
            switch ($action) {
                case 'create':
                    $result = $this->handleCreateCustom($data);
                    break;
                case 'update':
                    $result = $this->handleUpdateCustom($data);
                    break;
                case 'delete':
                    $result = $this->handleDeleteCustom($data);
                    break;
                default:
                    Yii::warning("Unknown custom action: {$action}", 'rabbitmq.custom');
                    $result = false;
            }
            
            if ($result) {
                Yii::info("Custom message processed successfully", 'rabbitmq.custom');
                return ConsumerInterface::MSG_ACK;
            } else {
                Yii::error("Custom message processing failed", 'rabbitmq.custom');
                return ConsumerInterface::MSG_REQUEUE;
            }
            
        } catch (\Exception $e) {
            Yii::error("Custom consumer exception: " . $e->getMessage(), 'rabbitmq.custom');
            return ConsumerInterface::MSG_REQUEUE;
        }
    }
    
    /**
     * 处理客户创建
     * 
     * @param array $data 消息数据
     * @return bool 处理结果
     */
    private function handleCreateCustom($data)
    {
        // TODO: 实现客户创建业务逻辑
        // 调用对应的 Service 或 Class 层处理
        return true;
    }
    
    /**
     * 处理客户更新
     * 
     * @param array $data 消息数据
     * @return bool 处理结果
     */
    private function handleUpdateCustom($data)
    {
        // TODO: 实现客户更新业务逻辑
        // 调用对应的 Service 或 Class 层处理
        return true;
    }
    
    /**
     * 处理客户删除
     * 
     * @param array $data 消息数据
     * @return bool 处理结果
     */
    private function handleDeleteCustom($data)
    {
        // TODO: 实现客户删除业务逻辑
        // 调用对应的 Service 或 Class 层处理
        return true;
    }
}