huahuibao/common/components/rabbitmq/baseConsumer.php

<?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;
    }
}