<?php
namespace KIF\Db;
use KIF\Core\Config;
use MySQLi as MySQLiResource;
use KIF\Debug\Debug;
use KIF\Core\Request;
use KIF\Cookie;
use Exception;
class MySQLi {
/**
* 当前的数据库连接对象
*
* @var MySQLiResource
*/
private $link;
/**
* 当前连接的数据库名
* @var string
*/
private $dbname;
/**
* #private
* 统计查询次数
*
* @var Int
*/
static public $queries = 0;
/**
* 通过$this->query 所执行的总时间
* 只有 debug::sql打开才会统计
*
* @var Int
*/
private static $intQueriesTotalTime = 0;
/**
* 保存所有已建立的到服务器的连接。
*
* @var Array 结构如下：
* array(
* '连接唯一标志' => MySQLiResource //连接唯一标志由 $this->getUniqueFlagOfLink() 获取；MySQLiResource 是到主服务器1的连接。
* , ...
* )
*/
static private $links;
/**
* 调试等级
* 0 不处理（交予外部程序处理）
* 1 显示错误并中断程序
* 2 直接中断程序
*
* @var Int
*/
public $debug_level = 1;
/**
* 最近一次执行的语句
*
* @var string
*/
private $last_sql;
/**
* 最后一次执行 query 后获取的 mysqli_result 对象
*
* @var mysqli_result
*/
private $result;
/**
* from Data Source Name (dsn)
* for example: mysqli://root:851031@localhost:3306/testDb?charset=utf8
*
* @var Array (
* 'prefix' => '',
* 'host' => '',
* 'port' => '',
* 'user' => '',
* 'pass' => '',
* 'params' => array(),// 通过dsn传递进来的查询字串
* )
*/
private $arrDsn;
/**
* 保存当前主服务器连接的信息
*是对 self::$masterDBInfos 数据结构某一项的引用
* @var array
* 结构如下
* array(// 当前集群下的主服务器
* 'host' => //主机名
* , 'port' => //端口
* , 'user' => //用户名
* , 'pass' => //密码
* , 'transactionIds' => //用于存储事务id。Array ('taId1' => true, 'taId2' => false, ...);
* //键代表事务标志，同时也是用于保存点的标志。值表示是否为顶层事务（只能有一个顶层事务）
* , 'isRunOnMaster' => Boolean //$this->query 方法里可检验该值，判断当前查询正连接于主服务器上
* , 'isUseMaster' => Boolean //指示后续查询是否要使用到主库的连接
* , 'arrStatusOfUseMaster' => Array(); //当嵌套修改 isUseMaster 的值时，需要有一个结构保存之前的状态，以便恢复
* )
*/
private $masterDBInfo;
/**
* 保存所有用于主服务器连接的信息
*
* @var Array 结构如下：
* array(
* '集群1标志' => array( //
* '主服务器标志' => array(//该标志通过 $this->getUniqueFlagOfLink() 获取:实质是通过 host, port, user, pass 来区别，所以同一集群如果使用不同的用户连接，会有不同的标志
* 'host' => //主机名
* , 'port' => //端口
* , 'user' => //用户名
* , 'pass' => //密码
* , 'transactionIds' => //用于存储事务id。Array ('taId1' => true, 'taId2' => false, ...);
* //键代表事务标志，同时也是用于保存点的标志。值表示是否为顶层事务（只能有一个顶层事务）
* , 'isRunOnMaster' => Boolean //$this->query 方法里可检验该值，判断$this->link是否正链接到主服务器上
* , 'isUseMaster' => Boolean //指示后续查询是否要使用到主库的连接
* , 'arrStatusOfUseMaster' => Array();
* )
* , ...
* )
* , ...
* )
*/
static private $masterDBInfos = array();
/**
* 集群信息，即用于候选的从服务器集
* array(
* 'clusterLevel' => (2 表示正常集群、3 表示低效集群)
* , 'Ignore_Check_Master_Patterns' => array()
*
* , '集群1标志' => array( // 该标志通过 $this->getMasterHost() 获取：集群只需要通过主服务器host即可区分
* array(
* 'host' =>
* , 'port' =>
* , 'user' =>
* , 'pass' =>
* )
* , ...
* )
* , ...
* )
* @var Array
*/
static private $slaveDBInfos;
/**
* 保存已从集群信息中获取的用于连接从服务器的信息
* @author Gxg <gaoxiaogang@gmail.com>
*
* @var Array 结构如下：
* array(// 当前集群下的从库信息
* 'host' => //主机名
* , 'port' => //端口
* , 'user' => //用户名
* , 'pass' => //密码
* )
*/
private $slaveDBInfo;
/**
* 返回当前集群中从服务器的信息
*
* @return false | Array 格式：
* array(
* 'host' => //主机名
* , 'user' => //用户名
* , 'pass' => //密码
* )
*/
private function getCurrentSlaveInfo() {
$arrInvalidSlaveInfos = $this->getCurrentInvalidSlaveInfos();
if (isset($this->slaveDBInfo)) {
# 检验是否有效
if (!$arrInvalidSlaveInfos || !in_array($this->slaveDBInfo['host'], $arrInvalidSlaveInfos)) {
return $this->slaveDBInfo;
}
}
$strUniqueFlagOfCluster = $this->getUniqueFlagOfCurrentCluster();
if (!isset(self::$slaveDBInfos[$strUniqueFlagOfCluster])) {
# 不能使用 $this->slaveDBInfo = null，否则引用的 self::$slaveDBInfos 相应的值也会变成 null
return false;
}
# 把 self::$slaveDBInfos 中已经无效的从库踢掉
if ($arrInvalidSlaveInfos) {
foreach (self::$slaveDBInfos[$strUniqueFlagOfCluster] as $key => $arrSlaveInfo) {
$strSlaveHost = $arrSlaveInfo['host'];
if (in_array($strSlaveHost, $arrInvalidSlaveInfos)) {
unset(self::$slaveDBInfos[$strUniqueFlagOfCluster][$key]);
}
}
}
if (count(self::$slaveDBInfos[$strUniqueFlagOfCluster]) == 0) {
return false;
}
$intRandPos = array_rand(self::$slaveDBInfos[$strUniqueFlagOfCluster]);
$this->slaveDBInfo = & self::$slaveDBInfos[$strUniqueFlagOfCluster][$intRandPos];
if (!isset($this->slaveDBInfo['user']) || !isset($this->slaveDBInfo['pass'])) {//从服务器没有提供连接帐户，则使用主服务器的帐户
$this->slaveDBInfo['user'] = $this->masterDBInfo['user'];
$this->slaveDBInfo['pass'] = $this->masterDBInfo['pass'];
}
if (!isset($this->slaveDBInfo['port'])) {
$this->slaveDBInfo['port'] = $this->masterDBInfo['port'];
}
return $this->slaveDBInfo;
}
/**
* 集群中无效的从服务器信息
* 寻找从服务器时，会忽略该结构中列出的主机
* array(
* '集群1标志' => array('从服务器host', '从服务器host', ...)
* , ...
* )
*
* @var Array
*/
static private $invalidSlaveInfos;
/**
* 增加无效的从服务器信息
*
* @param String $strHost
* @return true
*/
private function addInvalidSlaveInfo($strHost) {
$strUniqueFlagOfCluster = $this->getUniqueFlagOfCurrentCluster();
if (!isset(self::$invalidSlaveInfos[$strUniqueFlagOfCluster])
|| !in_array($strHost, self::$invalidSlaveInfos[$strUniqueFlagOfCluster])
) {
self::$invalidSlaveInfos[$strUniqueFlagOfCluster][] = $strHost;
}
return true;
}
/**
* 获取当前集群的唯一标志（即主服务器host）
*
* @return String
*/
private function getUniqueFlagOfCurrentCluster() {
//return "{$this->masterDBInfo['host']}_{$this->masterDBInfo['port']}_{$this->masterDBInfo['user']}";
return "{$this->masterDBInfo['host']}:{$this->masterDBInfo['port']}";
}
/**
* 获取当前集群下无效的从服务器信息
*
* @return false | Array
*/
private function getCurrentInvalidSlaveInfos() {
$strUniqueFlagOfCluster = $this->getUniqueFlagOfCurrentCluster();
if (!isset(self::$invalidSlaveInfos[$strUniqueFlagOfCluster])) {
return false;
}
return self::$invalidSlaveInfos[$strUniqueFlagOfCluster];
}
/**
* 设置客户端连接字符集
*
* @param String $charset 连接编码。该参数决定了从各种编码的表里取出数据后，以$strCharset指定的编码返回
* 比如：gbk编码存储的表数据，可以通过指定$strCharset为utf8来返回utf8编码的数据
* @return Boolean true:成功;false:失败.
*/
public function setCharset($charset = null) {
if (is_null($charset)) {
if (isset($this->arrDsn['params']['charset']) && $this->arrDsn['params']['charset']) {
$charset = $this->arrDsn['params']['charset'];
} else {
$charset = "utf8";
}
}
return $boolResult = mysqli_set_charset($this->link, $charset);
}
/**
* 取得客户端连接字符集
*
* @return String
*/
public function getCharset() {
return $strResult = mysqli_character_set_name($this->link);
}
/**
* 获取当前连接的唯一标志
*
* @return String
*/
private function getUniqueFlagOfLink() {
return "{$this->arrDsn['host']}_{$this->arrDsn['port']}_{$this->arrDsn['user']}_{$this->arrDsn['pass']}";
}
/**
* 构造函数
*
* @param String $dsn For Example: mysqli://root:851031@localhost/testDb
*/
public function __construct($dsn) {
$arrUrlInfo = parse_url($dsn);
if (!is_array($arrUrlInfo) || !isset($arrUrlInfo['host']) || !isset($arrUrlInfo['user'])
|| !isset($arrUrlInfo['pass']) || !isset($arrUrlInfo['path']))
{
$this->_halt("构造参数不正确：{$dsn}");
return false;
//解析出错时的处理
}
$this->arrDsn['host'] = $arrUrlInfo['host'];
$this->arrDsn['user'] = $arrUrlInfo['user'];
$this->arrDsn['pass'] = $arrUrlInfo['pass'];
$dbname = substr($arrUrlInfo['path'], 1);
if (empty($dbname)) {
return $this->_halt('请先设置数据库名', '21');
}
$this->dbname = $dbname;
isset($arrUrlInfo['scheme']) && $this->arrDsn['prefix'] = $arrUrlInfo['scheme'];
if (!isset($arrUrlInfo['port'])) {
$this->arrDsn['port'] = 3306;
} else {
$this->arrDsn['port'] = $arrUrlInfo['port'];
}
# 处理参数
if (isset($arrUrlInfo['query'])) {
parse_str($arrUrlInfo['query'], $this->arrDsn['params']);
} else {
$this->arrDsn['params'] = array();
}
$strUniqueFlagOfCluster = $this->arrDsn['host'];
$strUniqueFlagOfMaster = $this->getUniqueFlagOfLink();
if (!isset(self::$masterDBInfos[$strUniqueFlagOfCluster][$strUniqueFlagOfMaster])) {
self::$masterDBInfos[$strUniqueFlagOfCluster][$strUniqueFlagOfMaster] = array(
'host' => $this->arrDsn['host']
, 'port' => $this->arrDsn['port']
, 'user' => $this->arrDsn['user']
, 'pass' => $this->arrDsn['pass']
, 'transactionIds' => null
, 'isRunOnMaster' => false
, 'isUseMaster' => false
, 'arrStatusOfUseMaster' => null
);
}
$this->masterDBInfo = & self::$masterDBInfos[$strUniqueFlagOfCluster][$strUniqueFlagOfMaster];
if (!isset(self::$slaveDBInfos)) {
$this->initSlaveDBInfos();
}
}
/**
* 初始化 self::$slaveDBInfos 数据结构
*
* dbslaves 的结构：
* array(
* array(
* 'master' => (string),//对应主库的host值
* 'mixed' => array(//正常从库
* 'mysqld-6.yoka.com',
* '192.168.0.150',
* ... ,
* ),
* 'delay_mixed' => array(//慢速从库，比如用于提供给爬虫、或提供给翻页的比较后面的页面，无需保证特别好的服务
* 'mysqld-12.yoka.com',
* array(
* 'host' => (string),//必须
* 'user' => (string),//必须
* 'pass' => (string),//必须
* 'port' => (int),//非必须
* ),
* ... ,
* ),
* )
* );
*
*/
private function initSlaveDBInfos() {
try {
$dbslavesConfig = Config::getInstance()->get('dbslaves');
} catch (Exception $e) {
return false;
}
if (!is_array($dbslavesConfig)) {
return false;
}
$db_cluster_maps = $dbslavesConfig;
if (isset($_SERVER['Cluster_User_Level']) && $_SERVER['Cluster_User_Level'] == 3) {
$cluster_level = 3;
$strType = 'delay_mixed';
} else {
$cluster_level = 2;
$strType = 'mixed';
}
foreach ($db_cluster_maps as $arrClusterInfo) {
if (!isset($arrClusterInfo['master']) || !isset($arrClusterInfo[$strType]) || !is_array($arrClusterInfo[$strType]) || (count($arrClusterInfo[$strType]) == 0)) {
# TODO 记录日志
continue;
}
foreach($arrClusterInfo[$strType] as $mixConnectInfo) {
if (is_string($mixConnectInfo)) {//类似 mysqld-6.verycd.com 的值
self::$slaveDBInfos[$arrClusterInfo['master']][] = array(
'host' => $mixConnectInfo
, 'port' => null
, 'user' => null
, 'pass' => null
);
} elseif (is_array($mixConnectInfo)) {//如果$_SERVER['DataBase_Cluster_Map'] 提供客启端的密码，请使用以下格式 !
if (isset($mixConnectInfo['host']) && isset($mixConnectInfo['user']) && isset($mixConnectInfo['pass'])) {
$tmpPort = isset($mixConnectInfo['port']) ? $mixConnectInfo['port'] : 3306;
self::$slaveDBInfos[$arrClusterInfo['master']][] = array(
'host' => $mixConnectInfo['host']
, 'port' => $tmpPort
, 'user' => $mixConnectInfo['user']
, 'pass' => $mixConnectInfo['pass']
);
} else {
// $this->_halt('该集群：' . $arrClusterInfo['master'] . '在$_SERVER[\'DataBase_Cluster_Map\']里提供的连接信息格式错误');
# TODO 记下错误日志
continue;
}
} else {
# TODO 记下错误日志
// $this->_halt('该集群：' . $arrClusterInfo['master'] . '在$_SERVER[\'DataBase_Cluster_Map\']里提供的连接信息格式错误');
continue;
}
}//end foreach
}//end foreach
}
/**
* 获取当前主连接唯一标志
*
* @return String
*/
private function getUniqueFlagOfCurrentMaster() {
return "{$this->masterDBInfo['host']}_{$this->masterDBInfo['port']}_{$this->masterDBInfo['user']}_{$this->masterDBInfo['pass']}";
}
/**
* 获取当前集群状态下到主服务器的连接
*
* @return mysqli | false 已有连接，返回该连接（即mysqli对象）；false：没有连接
*/
private function getCurrentMasterLink() {
$strUniqueFlagOfCurrentMaster = $this->getUniqueFlagOfCurrentMaster();
if (isset(self::$links[$strUniqueFlagOfCurrentMaster])) {
if ($this->isLink(self::$links[$strUniqueFlagOfCurrentMaster])) {
return self::$links[$strUniqueFlagOfCurrentMaster];
} else {
# 曾建立过连接，但中途该连接失效了，应该对此做出处理的。
if (!empty($this->masterDBInfo['transactionIds'])) {
return $this->_halt("到主库的连接已失效，且该主库上存在事务，必须中断！");
}
}
}
return false;
}
/**
* 当前连接是否连到主库
*/
private function isRunOnMaster() {
return (boolean) $this->masterDBInfo['isRunOnMaster'];
}
/**
* 获取所有事务信息
* 只有当前集群中主服务器的连接是可用的，才有事务可言
*
* @return false | Array() false：没有到主服务器的连接或者还没开启事务；
*/
private function getTransactions() {
if ($this->getCurrentMasterLink()) {
if (!empty($this->masterDBInfo['transactionIds'])) {
return $this->masterDBInfo['transactionIds'];
}
}
return false;
}
/**
* 当前查询是否正运行在主服务器上并且开启了事务
* 该方法在 $this->query 方法里调用才是最有价值的。
*
* @return Boolean true：是；false：否
*/
private function isRunOnTransaction() {
if ($this->isRunOnMaster() && $this->getTransactions()) {
return true;
}
return false;
}
private function isLink($link) {
if (!($link instanceof MySQLiResource)) return false;
$sinfo = @mysqli_get_host_info($link);
return !empty($sinfo);
}
private function isReadSql($sql) {
static $r_ops = array('select','show','desc');
$sql = strtolower(trim($sql));
foreach ($r_ops as $op) {
if (strpos($sql,$op)===0) return true;
}
return false;
}
/**
* 连接数据库。这里创建的是真实链接，不会重用已有的链接
*
* @return false | MySQLiResource false：连接失败
*/
private function connect() {
// 连接数据库服务器
$objMysqli = mysqli_init();
$connect_rs = mysqli_real_connect($objMysqli, $this->arrDsn['host'], $this->arrDsn['user'], $this->arrDsn['pass']
, null, $this->arrDsn['port'], null
, MYSQLI_CLIENT_COMPRESS);
if (!$connect_rs) {
return false;
}
# 设置字符集的代码
$this->link = $objMysqli;
$this->setCharset();
return $this->link;
}
/**
* 标志当前连接$this->link连接到主库
*
* @return Boolean
*/
private function beginRunOnMaster() {
$this->masterDBInfo['isRunOnMaster'] = true;
return true;
}
/**
* 标志当前连接$this->link离开主库
*
* @return Boolean
*/
private function endRunOnMaster() {
$this->masterDBInfo['isRunOnMaster'] = false;
return true;
}
/**
* 该方法只应该由 $this->xconnect()调用
*
* @return Boolean false：连接失败
*/
private function connectMaster() {
# These codes increase by Gxg <gaoxiaogang@gmail.com>
# 保证到当前集群的主服务器的连接唯一
$this->arrDsn['host'] = $this->masterDBInfo['host'];
$this->arrDsn['port'] = $this->masterDBInfo['port'];
$this->arrDsn['user'] = $this->masterDBInfo['user'];
$this->arrDsn['pass'] = $this->masterDBInfo['pass'];
$objCurrentMasterLink = $this->getCurrentMasterLink();
if ($objCurrentMasterLink) {
$this->link = $objCurrentMasterLink;
} else {
if (false === $this->connect()) {
return false;
}
$strUniqueFlagOfLink = $this->getUniqueFlagOfLink();
self::$links[$strUniqueFlagOfLink] = $this->link;
}
$this->beginRunOnMaster();
return true;
}
/**
* 该方法只应该由 $this->xconnect()调用
*
* @return Boolean false：连接失败
*
*/
private function connectSlave() {
$arrCurrentSlaveInfo = $this->getCurrentSlaveInfo();
if ($arrCurrentSlaveInfo) {
$this->arrDsn['host'] = $arrCurrentSlaveInfo['host'];
$this->arrDsn['port'] = $arrCurrentSlaveInfo['port'];
$this->arrDsn['user'] = $arrCurrentSlaveInfo['user'];
$this->arrDsn['pass'] = $arrCurrentSlaveInfo['pass'];
$strUniqueFlagOfLink = $this->getUniqueFlagOfLink();
if (isset(self::$links[$strUniqueFlagOfLink]) && $this->isLink(self::$links[$strUniqueFlagOfLink])) {
$this->link = self::$links[$strUniqueFlagOfLink];
} else {
if(false === $this->connect()) {
return false;
}
self::$links[$strUniqueFlagOfLink] = $this->link;
}
return true;
} else {//还是连主服务器
return $this->connectMaster();
}
}
/**
*
* 重新链接。
* 目前在 mysql 2006错误，并且没有运行事务时才开启该方法
* @return Boolean
*/
private function reConnect() {
if (false === $this->connect()) {
return false;
}
$strUniqueFlagOfLink = $this->getUniqueFlagOfLink();
self::$links[$strUniqueFlagOfLink] = $this->link;
return true;
}
/**
* 判断查询是否要使用主服务器
*
* @return Boolean true：使用主；false：使用从
*/
private function isUseMaster() {
return (boolean) $this->masterDBInfo['isUseMaster'];
}
/**
* 保存当前状态，并置为$status
*
* @param Boolean $status
* @return String
*/
private function changeStatusOfUseMaster($status) {
($status === true) || $status = false;
# 保存当前状态
$strMasterStatusId = $this->getUniqueMasterStatusId();
$this->masterDBInfo['arrStatusOfUseMaster'][$strMasterStatusId] = $this->masterDBInfo['isUseMaster'];
$this->masterDBInfo['isUseMaster'] = $status;
return $strMasterStatusId;
}
/**
* 开始使用主库
*
* @return String 返回一串标志，供$this->restore 方法使用，用于恢复上一个状态
*/
public function beginUseMaster() {
return $this->changeStatusOfUseMaster(true);
}
/**
* 恢复采用 $strMasterStatusId 为句柄保存的上次的状态
*
* @param String $strMasterStatusId
* @return Boolean
*
*/
public function restore($strMasterStatusId) {
# 恢复指定状态
if (isset($this->masterDBInfo['arrStatusOfUseMaster'][$strMasterStatusId])) {
$this->masterDBInfo['isUseMaster'] = $this->masterDBInfo['arrStatusOfUseMaster'][$strMasterStatusId];
unset($this->masterDBInfo['arrStatusOfUseMaster'][$strMasterStatusId]);
return true;
}
return false;
}
/**
* 开始使用从库
* 尽量不要使用该接口，除非你明白自己真的需要
*
*/
public function beginUseSlave() {
return $this->changeStatusOfUseMaster(false);
}
/**
* 处理连接
*
* @param String $sql
*/
protected function xconnect($sql) {
$isUseSlave = $this->isReadSql($sql) && !$this->isUseMaster() ? true : false;
$intConnectErrorNum = 0;//连接出错次数
while(true) {
if ($isUseSlave) {
$isConnect = $this->connectSlave();
} else {
$isConnect = $this->connectMaster();
}
if (!$isConnect) {//连接失败
++$intConnectErrorNum;
$strMasterHost = $this->masterDBInfo['host'];
if(4 >= $intConnectErrorNum //允许四次重试
&& $this->arrDsn['host'] != $strMasterHost //错误不是发生在主服务器上
) {
$this->addInvalidSlaveInfo($this->arrDsn['host']);
// $this->addErrorLog(self::PARAM_NO_IMPORTANCE_ERROR_DIR, $intConnectErrorNum);
continue;
}
return $this->_halt('服务器连接失败', '01');
}
# 成功就退出循环
break;
}
}
/**
* 执行一条SQL
*
* @param String $sql
* @return resource result
*/
public function query($sql) {
# 每条语句都添加注释，方便debug。比如慢查询日志里知道问题出在哪个文件。
$sql .= '/* ' . $_SERVER['HTTP_HOST'] . ' in '.$_SERVER['PHP_SELF'] . ' */';
if (Debug::$open && preg_match('#^\s*select\s#i', $sql)) {
$explain_query = true;
} else {
$explain_query = false;
}
if ($explain_query) {
# 便于查看不使用缓存时的情况
$sql = preg_replace('#select #i', 'select sql_no_cache ', $sql);
}
$this->last_sql = $sql; // 临时加上
$intQueryErrorNum = 0;//查询出错次数
$intSelectErrorNum = 0;//选择数据库出错次数
while (true) {
$this->xconnect($sql);
$isSelect = mysqli_select_db($this->link, $this->dbname);
# 处理选择数据库错误
if (!$isSelect) {
if ($this->isRunOnTransaction()) {
return $this->_halt('进入数据库失败：存在事务，直接停机', '02');
}
++$intSelectErrorNum;
if ($intSelectErrorNum > 4) {
return $this->_halt('进入数据库失败后，重试多次后仍然失败', '02');
}
// 服务器链接丢失的错误，并且没有运行事务，允许重连
if ($this->errno() == 2006) {
if ($this->reConnect()) {
continue;
}
}
if ($this->isRunOnMaster()) {// 主库重连错误，停机
return $this->_halt('进入数据库失败', '02');
}
// 否则将这台重库置为无效，尝试重连
$this->addInvalidSlaveInfo($this->arrDsn['host']);
continue;
}
if (!$this->isReadSql($sql)) {
// # 如果是写入语句，记录开始时间
// $objProcessTimeOfWriteSqlTime = new ProcessTime();
// $objProcessTimeOfWriteSqlTime->start();
# 如果运行于事务中，记录该语句
if ($this->isRunOnTransaction()) {
$this->masterDBInfo['arrTransactionSqls'][] = $sql;
}
}
$query = mysqli_query($this->link, $sql);
// # 记录慢写入语句
// if (!$this->isReadSql($sql)) {
// if (($runTimeOfWriteSql = $objProcessTimeOfWriteSqlTime->getFinalTime()) > 1) {
// $this->addSlowWriteSqlLog($runTimeOfWriteSql);
// }
// }
# 处理查询错误
if (!$query) {
if ($this->isRunOnTransaction()) {
return $this->_halt('查询数据库失败：存在事务，直接停机', '21');
}
++$intQueryErrorNum;
if ($intQueryErrorNum > 4) {
return $this->_halt('查询数据库失败后，重试多次后仍然失败', '21');
}
// 服务器链接丢失的错误，并且没有运行事务，允许重连
if (in_array($this->errno(), array(2006, 2013))) {
if ($this->reConnect()) {
continue;
}
}
static $arrConnectErrnos = array(
1053 //在操作过程中服务器关闭。
, 1030 //从存储引擎中获得错误
, 126 //表损坏
);
if ($this->isRunOnMaster()) {// 主库重连错误，停机
return $this->_halt('查询主数据库失败', '21');
}
// 否则将这台重库置为无效，尝试重连
if(in_array($this->errno(), $arrConnectErrnos)) {//指定的错误号
$this->addInvalidSlaveInfo($this->arrDsn['host']);//由于连接失效导致的查询出错，将这台服务器标记为无效
continue;
}
return $this->_halt('查询数据库失败，不能处理的错误类型', '21');
}
break;
}
# 走到这里，说明成功的执行了写入sql
if ($this->canSetCookieForMasterDBHasWrite($sql)) {// 如果写入语句成功，就写一个保存时间为$tmp_expiration秒的cookie
$tmp_expiration = 2*60;
Cookie::set(KIF_MASTER_DB_HAS_WRITE_COOKIE_KEY, '1', $tmp_expiration);// 设置$tmp_expiration秒的cookie
}
self::$queries++;
if ($explain_query) {
$begin_microtime = Debug::getTime();
self::$intQueriesTotalTime += $begin_microtime;
$explainSql = 'explain ' . $sql;
$equery = mysqli_query($this->link, $explainSql);
$explain = $this->fetch($equery);
$this->freeResult($equery);
Debug::db($this->getLinkDesc(), $this->dbname, $explainSql, Debug::getTime() - $begin_microtime, $explain);
}
if (!$this->isReadSql($sql)) {
$begin_microtime = Debug::getTime();
Debug::db($this->getLinkDesc(), $this->dbname, $sql, Debug::getTime() - $begin_microtime, $query);
}
$this->isRunOnMaster() && $this->endRunOnMaster();
return $query;
}
/**
*
* 判断指定的sql执行后，能否设置后续查询转到主库的cookie
* @param string $sql
* @return boolean
*/
private function canSetCookieForMasterDBHasWrite($sql) {
if (Request::isCLI()) {
return false;
}
if ($this->isReadSql($sql)) {
return false;
}
# xhprof_logs表是用来记录慢查询的，没必要因为这个表写入了一次就把后续的请求都转到主库。
if (preg_match('#(xhprof_logs)#', strtolower($sql))) {
return false;
}
return true;
}
public function fetchOne($sql) {
$begin_microtime = Debug::getTime();
$res = $this->query($sql);
if (!$res) {
return false;
}
$result = $this->fetch($res);
$this->freeResult($res);
Debug::db($this->getLinkDesc(), $this->dbname, $this->last_sql, Debug::getTime() - $begin_microtime, $result);
return $result;
}
/**
* 获取连接描述，用于Debug输出
*/
private function getLinkDesc() {
$thread_id = mysqli_thread_id($this->link);
return "mysqli://{$this->arrDsn['user']}:{$this->arrDsn['port']}@{$this->arrDsn['host']} (thread_id: {$thread_id})";
}
/**
* 执行一条SQL并返回此查询包含的所有数据(2维数组)
*
* @param string $sql
* @param string $associateKey 如果指定了$associateKey，返回结果以 每条记录的$associateKey字段做数组下标
* @return false | array
*/
public function fetchAll($sql, $associateKey = null) {
$begin_microtime = Debug::getTime();
$res = $this->query($sql);
if (!$res) {
return false;
}
$result = array();
if ($associateKey) {
while (true) {
$row = $this->fetch($res);
if (!$row) {
break;
}
if (isset($row[$associateKey])) {
$result[$row[$associateKey]] = $row;
} else {
$result[] = $row;
}
}
} else {
while (true) {
$row = $this->fetch($res);
if (!$row) {
break;
}
$result[] = $row;
}
}
$this->freeResult($res);
Debug::db($this->getLinkDesc(), $this->dbname, $this->last_sql, Debug::getTime() - $begin_microtime, $result);
return $result;
}
/**
* 执行SQL语句并返回第一行第一列
*
* @param string $sql
* @return false | scala
*/
public function fetchSclare($sql) {
$begin_microtime = Debug::getTime();
$result = $this->fetchOne($sql);
if (!$result) {
return false;
}
$result = array_shift($result);
Debug::db($this->getLinkDesc(), $this->dbname, $this->last_sql, Debug::getTime() - $begin_microtime, $result);
return $result;
}
/**
* 返回上一步 INSERT 查询中产生的 AUTO_INCREMENT 的 ID 号；
* 或者 返回 update 语句中 last_insert_id()函数中表达式的值。
*
* ！！请记住，一定紧接在insert 或 update 语句后执行该方法，否则$this->link可能已经指向别的服务器了
*
* @return int | NULL >0：成功取到；0：没取到；NULL：$this->link无效
*/
public function insertId() {
return mysqli_insert_id($this->link);
}
/**
* $this->insertId() 的别名
* ！！请记住，一定紧接在insert 或 update 语句后执行该方法，否则$this->link可能已经指向别的服务器了
* @return int | NULL >0：成功取到；0：没取到；NULL：$this->link无效
*/
public function getLastInsertId() {
return $this->insertId();
}
/**
* 返回最近一次 INSERT，UPDATE 或 DELETE 查询所影响的记录行数。
* @return int | null 返回值 >= 0：成功；等于 -1：最后一条查询错误；null：$this->link无效
**/
public function affectedRows()
{
return mysqli_affected_rows($this->link);
}
public function fetch($query, $resulttype = MYSQLI_ASSOC) {
return mysqli_fetch_array($query, $resulttype);
}
protected function freeResult($query) {
return mysqli_free_result($query);
}
/**
* 生成唯一的字符串作为事务的唯一id
*
* @return String
*/
static private function getUniqueTransactionId() {
return self::getUniqueId('TAId');
}
/**
* 生成唯一的字符串作为保存当前主服务器状态的唯一id
*
* @return String
*/
static private function getUniqueMasterStatusId() {
return self::getUniqueId('MSId');
}
/**
* 生成唯一id
*
* @param String $prefix
* @return String
*/
static private function getUniqueId($prefix = '') {
if(!is_string($prefix)) {
$prefix = '';
}
return uniqid($prefix . '_'.rand());
}
/**
* 返回上一个错误文本，如果没有出错则返回 ''（空字符串）。
* 如果没有指定连接资源号，则使用上一个成功打开的连接从数据库服务器提取错误信息。
*
* @return String
*/
public function error() {
return @mysqli_error($this->link);
}
/**
* 返回上一个错误号
*
* @return int | NULL
*/
public function errno() {
return @mysqli_errno($this->link);
}
/**
* 返回上一个连接错误
*
* @return String
*/
public function connect_error() {
return @mysqli_connect_error();
}
/**
* 返回上一个连接的错误号
*
* @return int
*/
public function connect_errno() {
return @mysqli_connect_errno();
}
/**
* 根据 $this->debug_level 处理一些异常情况
* 1 直接输出错误信息并中断程序
* 2 直接中断程序
* 其他情况不处理错误，返回flase，修改错误代号和本函数所提供的错误信息，最后的是MySQL服务器提供的信息
*
* @param String $msg
* @param String $errorcode
* @return Array
*/
function _halt($msg, $errorcode = '00') {
switch ($this->debug_level) {
case 1:
ob_clean();
header("HTTP/1.0 500 Server Error");
header("Expires: ".gmdate("D, d M Y H:i:s", time())." GMT");
header("Last-Modified: ".gmdate("D, d M Y H:i:s", time())." GMT");
header("Cache-Control: private");
// $out = file_get_contents(ROOT_PATH . '/mysql.html');
$out = '$the_error';
$out = str_replace('$the_error', $msg.'<hr />'.$this->error().' No.'.$this->errno()."<!-- {$this->last_sql} -->", $out);
// $this->addErrorLog(null, null, $msg);
echo $out;
exit;
break;
case 2:
ob_clean();
header("HTTP/1.0 500 Server Error");
header("Expires: ".gmdate("D, d M Y H:i:s", time())." GMT");
header("Last-Modified: ".gmdate("D, d M Y H:i:s", time())." GMT");
header("Cache-Control: private");
echo $this->connect_error(), "<br />";
echo $this->connect_errno(), "<br />";
echo "{$msg}<br />";
exit('MySQL.');
break;
default:
$this->errorcode = array($errorcode, $msg.':'.$this->last_sql, $this->errno().": ".$this->error());
return false;
break;
}
}
########### TRANSACTION ##########
/**
* 开启事务
*
* @return false | string false：失败；string：成功返回事务标志
*/
public function startTransaction() {
$strTransactionId = self::getUniqueTransactionId();
if ($this->getTransactions()) {//已存在事务
if ($this->setSavePoint($strTransactionId)) {
$this->masterDBInfo['transactionIds'][$strTransactionId] = false;
return $strTransactionId;
}
} else {
# 初次开启事务
if (true === $this->query('START TRANSACTION;')) {//如果没有建立到当前主服务器的连接，该操作会隐式的建立
$this->masterDBInfo['transactionIds'][$strTransactionId] = true;
return $strTransactionId;
}
}
# 开启事务失败。返回一个独特的字符串，该字符串是不可能出现在 事务id数组中的
return false;
}
/**
* 回滚父事务
*
* @param String $strTransactionId
* @return Boolean
*/
private function _rollbackRootTransaction($strTransactionId) {
if ($this->isRootTransaction($strTransactionId)) {//父事务
$this->masterDBInfo['transactionIds'] = null;
$this->masterDBInfo['arrTransactionSqls'] = array();
return $this->query('ROLLBACK;');
}
return false;
}
/**
* 回滚子事务
*
* @param String $strTransactionId
* @return Boolean
*/
private function _rollbackSubTransaction($strTransactionId) {
if($this->isSubTransaction($strTransactionId)) {//子事务
$boolStatusTmp = $this->rollbackToSavePoint($strTransactionId);
$this->releaseSavePoint($strTransactionId);
unset($this->masterDBInfo['transactionIds'][$strTransactionId]);
return $boolStatusTmp;
}
return false;
}
/**
* 撤消指定事务
*
* @param String $strTransactionId
* @return Bollean true:成功；false:失败
*/
public function rollback($strTransactionId) {
if ($this->isRootTransaction($strTransactionId)) {//父事务
return $this->_rollbackRootTransaction($strTransactionId);
} elseif ($this->isSubTransaction($strTransactionId)) {//子事务
return $this->_rollbackSubTransaction($strTransactionId);
} else {
return false;
}
}
/**
* 提交父事务
*
* @param String $strTransactionId
* @return Boolean
*/
private function _commitRootTransaction($strTransactionId) {
if ($this->isRootTransaction($strTransactionId)) {//父事务
$this->masterDBInfo['transactionIds'] = null;
$this->masterDBInfo['arrTransactionSqls'] = array();
return $this->query('COMMIT;');
}
return false;
}
/**
* 提交子事务
*
* @param String $strTransactionId
* @return Boolean
*/
private function _commitSubTransaction($strTransactionId) {
if ($this->isSubTransaction($strTransactionId)) {//子事务
$this->releaseSavePoint($strTransactionId);
unset($this->masterDBInfo['transactionIds'][$strTransactionId]);
return true;
}
return false;
}
/**
* 提交指定事务
*
* @param String $strTransactionId
* @return Boolean true:成功；false:失败
*/
public function commit($strTransactionId) {
if ($this->isRootTransaction($strTransactionId)) {//父事务
return $this->_commitRootTransaction($strTransactionId);
} elseif ($this->isSubTransaction($strTransactionId)) {//子事务
return $this->_commitSubTransaction($strTransactionId);
} else {
return false;
}
}
/**
* 设置子事务的保存点，用于支持子事务的回滚
*
* @param String $SPId 应该被传递的值是事务的唯一id，即调用self::getUniqueTransactionId()生成的
* @return Boolean true:成功；false:失败
*/
private function setSavePoint($SPId) {
if (true === $this->query("SAVEPOINT {$SPId}")) {
return true;
}
return false;
}
/**
* 获取指定事务的类型（父事务 还是 子事务）
* 只有当前集群中主服务器的连接是可用的，才有事务可言
*
* @param String $strTransactionId
* @return Boolean | null true：父事务；false：子事务；null：无效
*/
private function getTransactionTypeById($strTransactionId) {
if ($this->getCurrentMasterLink()) {
if (isset($this->masterDBInfo['transactionIds'][$strTransactionId])) {
return $this->masterDBInfo['transactionIds'][$strTransactionId];
}
}
return null;
}
/**
* 是否父事务
*
* @param String $strTransactionId
* @return Boolean true：是；false：否
*/
private function isRootTransaction($strTransactionId) {
if (true === $this->getTransactionTypeById($strTransactionId)) {
return true;
}
return false;
}
/**
* 是否子事务
*
* @param String $strTransactionId
* @return Boolean true：是；false：否
*/
private function isSubTransaction($strTransactionId) {
if (false === $this->getTransactionTypeById($strTransactionId)) {
return true;
}
return false;
}
/**
* 回滚到指定事务点
*
* @param String $SPId
* @return Boolean true:成功；false:失败
*/
private function rollbackToSavePoint($SPId) {
# 只对子事务的回滚点操作
if ($this->isSubTransaction($SPId)) {
if (true === $this->query("ROLLBACK TO SAVEPOINT {$SPId}")) {
return true;
}
}
return false;
}
/**
* 释放事务保存点
*
* @param String $SPId
* @return Boolean true:成功；false:失败
*/
private function releaseSavePoint($SPId) {
# 只对子事务的回滚点操作
if ($this->isSubTransaction($SPId)) {
if (true === $this->query("RELEASE SAVEPOINT {$SPId}")) {
return true;
}
}
return false;
}
}