sphinx/Frame/Library/MySQL.php

<?php
namespace PhpLife\Frame\Library;

use PhpLife\Frame\Logger;
use PhpLife\Frame\Exception;

/**
 * 处理MySQL连接、查询和写入
 *
 * @package PhpLife\Frame\Library
 */
class MySQL extends \PhpLife\Frame\Library
{

    private static $throwException = false;
    // 出错后是否抛出异常
    private static $inTransaction = false;
    // 是否在事务中
    private static $newConnection = false;
    // 是否使用新连接
    const CONNECTION_TIMEOUT = 21600;
    // 每个连接最多可以使用时间，超过这个时间之后创建新连接
    protected $exception;

    /**
     * 执行查询操作
     *
     * @param string $sql            
     * @param array $data            
     * @param bool $isMaster            
     * @param bool $fetchAll            
     * @throws \Exception
     * @return array|mixed
     */
    public function read($sql, $data, $isMaster = false, $fetchAll = true)
    {
        try {
            $stmt = $this->getConnection($isMaster)->prepare($sql);
            $stmt->execute($data);
            $stmt->setFetchMode(\PDO::FETCH_ASSOC);
            if ($fetchAll) {
                $result = $stmt->fetchAll();
            } else {
                $result = $stmt->fetch();
            }
        } catch (\Exception $ex) {
            $result = null;
            $this->exception = $ex;
            $message = array(
                'info' => $ex->getCode() . '=>' . $ex->getMessage() . ';' . json_encode($ex->errorInfo),
                'sql' => $sql,
                'data' => $data
            );
            Logger::error($message, 'mysql.read.error');
            // 如果是失去连接的错误,则进行重试
            if (in_array($ex->errorInfo[1], array(
                '2006',
                '2013'
            ))) {
                self::$newConnection = true;
                return $this->read($sql, $data, $isMaster, $fetchAll);
            }
            // 处理事务时抛出异常
            if (self::$throwException) {
                throw new \Exception($ex->getCode() . '=>' . $ex->getMessage(), '90203');
            }
        }
        Logger::debug(array(
            'sql' => $sql,
            'data' => $data,
            'result' => $result
        ), 'mysql.read.debug');
        return $result;
    }

    /**
     * 执行写入操作
     *
     * @param string $sql            
     * @param
     *            $data
     * @throws \Exception
     * @return boolean
     */
    public function write($sql, $data)
    {
        try {
            $stmt = $this->getConnection(true)->prepare($sql);
            $stmt->execute($data);
            $result = $stmt->rowCount();
        } catch (\Exception $ex) {
            $result = false;
            $this->exception = $ex;
            $message = array(
                'info' => $ex->getCode() . '=>' . $ex->getMessage() . ';' . json_encode($ex->errorInfo),
                'sql' => $sql,
                'data' => $data
            );
            Logger::error($message, 'mysql.write.error');
            // 如果是失去连接的错误,则进行重试
            if (in_array($ex->errorInfo[1], array(
                '2006',
                '2013'
            ))) {
                self::$newConnection = true;
                return $this->write($sql, $data);
            }
            // 处理事务时抛出异常
            if (self::$throwException) {
                throw new \Exception($ex->getCode() . '=>' . $ex->getMessage(), '90202');
            }
        }
        Logger::debug(array(
            'sql' => $sql,
            'data' => $data,
            'result' => $result
        ), 'mysql.write.debug');
        return $result;
    }

    /**
     * 事务处理, $callback中如果出错必需抛出异常,否则会真正执行
     * 开始时会设置为主动抛出异常,结束时会设置为关闭抛出异常
     *
     * @param
     *            $callback
     * @param
     *            $throwException
     * @return bool
     */
    public function transaction($callback, $throwException = false)
    {
        self::$inTransaction = true; // 处于事务中
        self::$throwException = $throwException; // 执行SQL出错时候是否抛出异常
        try {
            $connection = $this->getConnection(true);
            $connection->beginTransaction();
            $callback();
            $connection->commit();
            $result = true;
        } catch (\Exception $ex) {
            $result = false;
            $message = array(
                'info' => $ex->getCode() . '=>' . $ex->getMessage() . ';' . json_encode($ex->errorInfo)
            );
            Logger::error($message, 'mysql.transaction.error');
            // 如果是失去连接的错误,则进行重试
            if (in_array($ex->errorInfo[1], array(
                '2006',
                '2013'
            ))) {
                self::$newConnection = true;
                return $this->transaction($callback, $throwException);
            }
         
            try {
                $this->getConnection(true)->rollBack();
               
            } catch (\Exception $e) {
                
                // rollback failed
            }
            if($throwException){
                throw new Exception($ex->getMessage(),$ex->getCode());
            }
        }
        self::$throwException = false; // 关闭抛出异常
        self::$inTransaction = false; // 事务结束
        Logger::debug(array(
            'result' => $result
        ), 'mysql.transaction.debug');
        return $result;
    }

    /**
     * 获取自增ID
     *
     * @return string
     */
    public function getLastInsertId()
    {
        return $this->getConnection(true)->lastInsertId();
    }

    /**
     * 获取异常
     *
     * @return mixed|\Exception
     */
    public function getLastException()
    {
        return $this->exception;
    }

    /**
     * 获取连接池
     * 如果超时时处于事务中,则不主动断开重连
     *
     * @param bool $isMaster            
     * @return \PDO
     */
    protected function getConnection($isMaster = false)
    {
        static $pool = array();
        $connectionName = $this->getFlag($isMaster);
        $connectionTimeout = $connectionName . '_timeout';
        if (self::$newConnection || ! isset($pool[$connectionName]) || $pool[$connectionName] == false || (self::$inTransaction == false && (time() - $pool[$connectionTimeout] > self::CONNECTION_TIMEOUT))) {
            $cfg = $this->getConfig();
            if ($isMaster) {
                $cfg = $cfg['master'];
            } else {
                $cfg = $cfg['slave_list'][array_rand($cfg['slave_list'])];
            }
            $pool[$connectionName] = $this->connect($cfg);
            $pool[$connectionTimeout] = time();
            self::$newConnection = false;
        }
        return $pool[$connectionName];
    }

    /**
     * 连接数据库
     *
     * @param
     *            $cfg
     * @return \PDO
     * @throws \Exception
     */
    protected function connect($cfg)
    {
        $opts = array(
            \PDO::ATTR_AUTOCOMMIT => true,
            \PDO::ATTR_ERRMODE => \PDO::ERRMODE_EXCEPTION,
            \PDO::ATTR_TIMEOUT => $cfg['timeout'],
            \PDO::MYSQL_ATTR_INIT_COMMAND => "SET NAMES '" . $cfg['charset'] . "'",
            \PDO::ATTR_PERSISTENT => $cfg['persistent']?true : false
        ); // 启用长连接
        
        $dsn = 'mysql:dbname=' . $cfg['dbname'] . ';host=' . $cfg['host'] . ';port=' . $cfg['port'];
        try {
            $db = new \PDO($dsn, $cfg['username'], $cfg['password'], $opts);
        } catch (\PDOException $ex) {
            $this->exception = $ex;
            $message = "MySQL connection failed [" . $dsn . ';' . $cfg['username'] . ';password]' . $ex->getCode() . '=>' . $ex->getMessage();
            Logger::emerg($message, 'mysql.connect.error');
            throw new \PDOException($message, 90201);
        }
        $message = "MySQL connect [" . $dsn . ';' . $cfg['username'] . ';password]';
        Logger::debug($message, 'mysql.connect.debug');
        return $db;
    }
}