rabbitMQ 消费者异常修复: RabbitMQ 长驻消费者拿着“空闲太久的 MySQL 连接”去执行 SQL，触发 2006 MySQL server has gone away

common/components/rabbitmq/baseConsumer.php

@@ -0,0 +1,76 @@
+<?php
+
+namespace common\components\rabbitmq;
+
+use mikemadisonweb\rabbitmq\components\ConsumerInterface;
+use common\components\noticeUtil;
+use Yii;
+
+abstract class baseConsumer implements ConsumerInterface
+{
+    /**
+     * 在长驻进程中保证 db 连接可用
+     */
+    protected function ensureDbConnection()
+    {
+        $db = Yii::$app->db;
+        try {
+            if (!$db->isActive) {
+                $db->open();
+                return;
+            }
+            $db->createCommand('SELECT 1')->queryScalar();
+        } catch (\Throwable $e) {
+            // 连接失效时主动重建
+            $db->close();
+            $db->open();
+        }
+    }
+
+    /**
+     * 遇到 MySQL 断连时自动重连并重试一次
+     *
+     * @param callable $callback
+     * @return mixed
+     * @throws \Throwable
+     */
+    protected function runWithDbReconnect(callable $callback)
+    {
+        try {
+            return $callback();
+        } catch (\Throwable $e) {
+            if (!$this->isMysqlConnectionLost($e)) {
+                throw $e;
+            }
+            Yii::warning('检测到 MySQL 连接断开，尝试重连并重试一次: ' . $e->getMessage(), __METHOD__);
+            noticeUtil::push('检测到 MySQL 连接断开，尝试重连并重试一次: ' . $e->getMessage(), __METHOD__);
+            Yii::$app->db->close();
+            Yii::$app->db->open();
+            return $callback();
+        }
+    }
+
+    /**
+     * 判断是否属于 MySQL 连接丢失错误
+     *
+     * @param \Throwable $e
+     * @return bool
+     */
+    protected function isMysqlConnectionLost(\Throwable $e)
+    {
+        $message = $e->getMessage();
+        if (stripos($message, 'server has gone away') !== false) {
+            return true;
+        }
+        if (stripos($message, 'Lost connection to MySQL server') !== false) {
+            return true;
+        }
+        if (stripos($message, 'SQLSTATE[HY000] [2006]') !== false) {
+            return true;
+        }
+        if (stripos($message, 'SQLSTATE[HY000] [2013]') !== false) {
+            return true;
+        }
+        return false;
+    }
+}

common/components/rabbitmq/cancelLimitBuyConsumer.php

@@ -12,7 +12,7 @@ use common\components\noticeUtil;
 use mikemadisonweb\rabbitmq\components\ConsumerInterface;
 use PhpAmqpLib\Message\AMQPMessage;
 
-class cancelLimitBuyConsumer implements ConsumerInterface
+class cancelLimitBuyConsumer extends baseConsumer
 {
     /**
      * 执行消费者逻辑
@@ -27,6 +27,7 @@ class cancelLimitBuyConsumer implements ConsumerInterface
     public function execute(AMQPMessage $msg)
     {
         try {
+            $this->ensureDbConnection();
             // 反序列化消息体
             $data = unserialize($msg->body);
             if (!is_array($data)) {
@@ -36,14 +37,15 @@ class cancelLimitBuyConsumer implements ConsumerInterface
             print_r($data);
             // 根据操作类型分发处理
             $type = $data['type'] ?? null;
-            switch ($type) {
-                case 'limit_buy_clear':
-                    $result = $this->clearOrderItemLimitBuy($data);
-                    break;
-                default:
-                    noticeUtil::push("取消限购的消费者报错，未知 type: {$type}");
-                    $result = false;
-            }
+            $result = $this->runWithDbReconnect(function () use ($type, $data) {
+                switch ($type) {
+                    case 'limit_buy_clear':
+                        return $this->clearOrderItemLimitBuy($data);
+                    default:
+                        noticeUtil::push("取消限购的消费者报错，未知 type: {$type}");
+                        return false;
+                }
+            });
             if ($result) {
                 return ConsumerInterface::MSG_ACK;
             } else {

common/components/rabbitmq/customConsumer.php

@@ -12,7 +12,7 @@ use common\components\noticeUtil;
 use mikemadisonweb\rabbitmq\components\ConsumerInterface;
 use PhpAmqpLib\Message\AMQPMessage;
 
-class customConsumer implements ConsumerInterface
+class customConsumer extends baseConsumer
 {
     /**
      * 执行消费者逻辑
@@ -27,6 +27,7 @@ class customConsumer implements ConsumerInterface
     public function execute(AMQPMessage $msg)
     {
         try {
+            $this->ensureDbConnection();
             // 反序列化消息体
             $data = unserialize($msg->body);
             if (!is_array($data)) {
@@ -36,23 +37,22 @@ class customConsumer implements ConsumerInterface
             print_r($data);
             // 根据操作类型分发处理
             $type = $data['type'] ?? null;
-            switch ($type) {
-                case 'add_custom':
-                    //添加新客户
-                    $result = CustomClass::generateCustom($data);
-                    break;
-                case 'pull_custom_from_other_shop':
-                    //从分店拉取客户
-                    $result = CustomClass::pullOtherShopCustom($data);
-                    break;
-                case 'hd_change_custom_expense_level':
-                    //客户消费等级变更
-                    $result = HdCustomClass::updateCustomExpenseLevel($data);
-                    break;
-                default:
-                    noticeUtil::push("客户的消费者报错，不存在的类型: {$type}");
-                    $result = false;
-            }
+            $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 {

common/components/rabbitmq/notifyConsumer.php

@@ -16,7 +16,7 @@ use common\components\push;
 use mikemadisonweb\rabbitmq\components\ConsumerInterface;
 use PhpAmqpLib\Message\AMQPMessage;
 
-class notifyConsumer implements ConsumerInterface
+class notifyConsumer extends baseConsumer
 {
     /**
      * 执行消费者逻辑
@@ -31,6 +31,7 @@ class notifyConsumer implements ConsumerInterface
     public function execute(AMQPMessage $msg)
     {
         try {
+            $this->ensureDbConnection();
             // 反序列化消息体
             $data = unserialize($msg->body);
             if (!is_array($data)) {
@@ -40,31 +41,28 @@ class notifyConsumer implements ConsumerInterface
             print_r($data);
             // 根据通知类型分发处理
             $type = $data['type'] ?? null;
-            switch ($type) {
-                case 'ghs_new_order_notify':
-                    //供货商的新订单通知
-                    $result = $this->ghsNewOrderNotify($data);
-                    break;
-                case 'hd_new_order_notify':
-                    //花店的新订单通知
-                    $result = $this->hdNewOrderNotify($data);
-                    break;
-                case 'hd_new_cg_notify':
-                    //花店的新采购单通知
-                    $result = $this->hdNewCgNotify($data);
-                    break;
-                case 'ghs_pt_error':
-                    //供货商跑腿异常通知
-                    $result = $this->ghsPtErrorAction($data);
-                    break;
-                case 'hd_pt_error':
-                    //花店跑腿异常通知
-                    $result = $this->hdPtErrorAction($data);
-                    break;
-                default:
-                    noticeUtil::push("通知的消费者提示：Unknown notify type: {$type}", '15280215347');
-                    $result = false;
-            }
+            $result = $this->runWithDbReconnect(function () use ($type, $data) {
+                switch ($type) {
+                    case 'ghs_new_order_notify':
+                        //供货商的新订单通知
+                        return $this->ghsNewOrderNotify($data);
+                    case 'hd_new_order_notify':
+                        //花店的新订单通知
+                        return $this->hdNewOrderNotify($data);
+                    case 'hd_new_cg_notify':
+                        //花店的新采购单通知
+                        return $this->hdNewCgNotify($data);
+                    case 'ghs_pt_error':
+                        //供货商跑腿异常通知
+                        return $this->ghsPtErrorAction($data);
+                    case 'hd_pt_error':
+                        //花店跑腿异常通知
+                        return $this->hdPtErrorAction($data);
+                    default:
+                        noticeUtil::push("通知的消费者提示：Unknown notify type: {$type}", '15280215347');
+                        return false;
+                }
+            });
             if ($result) {
                 return ConsumerInterface::MSG_ACK;
             } else {

common/components/rabbitmq/ptConsumer.php

@@ -8,7 +8,7 @@ use common\components\noticeUtil;
 use mikemadisonweb\rabbitmq\components\ConsumerInterface;
 use PhpAmqpLib\Message\AMQPMessage;
 
-class ptConsumer implements ConsumerInterface
+class ptConsumer extends baseConsumer
 {
     /**
      * 执行消费者逻辑
@@ -23,6 +23,7 @@ class ptConsumer implements ConsumerInterface
     public function execute(AMQPMessage $msg)
     {
         try {
+            $this->ensureDbConnection();
             // 反序列化消息体
             $data = unserialize($msg->body);
             if (!is_array($data)) {
@@ -33,19 +34,19 @@ class ptConsumer implements ConsumerInterface
 
             // 根据操作类型分发处理
             $type = $data['type'] ?? null;
-            switch ($type) {
-                case 'hd_pt_create_order':
-                    //花店跑腿开始下单
-                    $result = \bizHd\express\classes\HdDeliveryOrderClass::beginCreateOrder($data);
-                    break;
-                case 'ghs_pt_create_order':
-                    //供货商跑腿开始下单
-                    $result = \bizGhs\express\classes\GhsDeliveryOrderClass::beginCreateOrder($data);
-                    break;
-                default:
-                    noticeUtil::push("跑腿发单的消费者报错，不存在的类型: {$type}");
-                    $result = false;
-            }
+            $result = $this->runWithDbReconnect(function () use ($type, $data) {
+                switch ($type) {
+                    case 'hd_pt_create_order':
+                        //花店跑腿开始下单
+                        return \bizHd\express\classes\HdDeliveryOrderClass::beginCreateOrder($data);
+                    case 'ghs_pt_create_order':
+                        //供货商跑腿开始下单
+                        return \bizGhs\express\classes\GhsDeliveryOrderClass::beginCreateOrder($data);
+                    default:
+                        noticeUtil::push("跑腿发单的消费者报错，不存在的类型: {$type}");
+                        return false;
+                }
+            });
 
             if ($result == ConsumerInterface::MSG_ACK) {
                 return ConsumerInterface::MSG_ACK;

common/components/rabbitmq/stockConsumer.php

@@ -12,7 +12,7 @@ use mikemadisonweb\rabbitmq\components\ConsumerInterface;
 use PhpAmqpLib\Message\AMQPMessage;
 use Yii;
 
-class stockConsumer implements ConsumerInterface
+class stockConsumer extends baseConsumer
 {
     /**
      * 执行消费者逻辑
@@ -27,6 +27,7 @@ class stockConsumer implements ConsumerInterface
     public function execute(AMQPMessage $msg)
     {
         try {
+            $this->ensureDbConnection();
             // 反序列化消息体
             $data = unserialize($msg->body);
             if (!is_array($data)) {
@@ -47,7 +48,9 @@ class stockConsumer implements ConsumerInterface
                     break;
                 case 'limit_buy_clear':
                     echo 'limit_buy_clear---';
-                    $result = $this->clearOrderItemLimitBuy($data);
+                    $result = $this->runWithDbReconnect(function () use ($data) {
+                        return $this->clearOrderItemLimitBuy($data);
+                    });
                     break;
                 default:
                     noticeUtil::push("库存的消费者报错，未知 type: {$type}");