mysql读写分离(PHP类)

博客分类：

• php
• mysql

自己实现了php的读写分离,并且不用修改程序

优点:实现了读写分离,不依赖服务器硬件配置,并且都是可以配置read服务器,无限扩展

缺点:错误转移不购明确,现只有3次尝试机会,失败了就会选择其他的服务器.当服务器恢复时,需要手动去消除错误日志不人性化

修改于discux x 的mysql类

配置文件

"db"=>array(
        '1'=>array(
            'dbhost'=> '127.0.0.1',
            'dbuser' => 'root',
            'dbpw' => 'xxx',
            'dbcharset' => 'utf8',
            'pconnect' => '0',
            'dbname' => 'test',
            'tablepre' => 'w_',
        ),
        '2'=>array(
            'dbhost'=> '127.0.0.1',
            'dbuser' => 'root',
            'dbpw' => 'xxx',
            'dbcharset' => 'utf8',
            'pconnect' => '0',
            'dbname' => 'test2',
            'tablepre' => 'w_',
        ),
        
        '3'=>array(
            'dbhost'=> '127.0.0.1',
            'dbuser' => 'root',
            'dbpw' => 'xxx',
            'dbcharset' => 'utf8',
            'pconnect' => '0',
            'dbname' => 'test3',
            'tablepre' => 'w_',
        ),
        'writes'=>array('2','3'),
        'reads'=>array('2','1'),
        'broken'=>array(),
        'broken_file'=>str_replace('\\','/',dirname(__FILE__)) ."/____mysql_broken_file.php",
        //'broken_file'=>array(),
     )

类文件

<?php

/*
    [Discuz!] (C)2001-2009 Comsenz Inc.
    This is NOT a freeware, use is subject to license terms

$Id: db_mysql.class.php 20294 2009-09-23 06:00:37Z zhaoxiongfei $
*/

if(!defined('IN_APP')) {
    exit('Access Denied');
}

class lib_db_mysql extends app_lib
{
    var $tablepre;
    var $version = '';
    var $querynum = 0;
    var $curlink;
    var $link = array();
    var $config = array();
    var $sqldebug = array();
    var $map = array();
    
    var $reads=array();//读服务器id
    var $writes=array();//写服务器id
    var $broken=array();//重试3次坏掉的服务器
    var $broken_file="";
    
    function init(){
        $this->set_config(config_item('db'));
    }

function set_config($config) {
        $this->config = &$config;
        $this->tablepre = $config['1']['tablepre'];
        if(!empty($this->config['map'])) {
            $this->map = $this->config['map'];
        }
        
        /**
        * 读写分离
        */
        if(!empty($this->config['reads'])) {
            $this->reads=$this->config['reads'];
            unset($this->config['reads']);
        }
        
        if(!empty($this->config['writes'])) {
            $this->writes=$this->config['writes'];
            unset($this->config['writes']);
        }

if(!empty($this->config['broken_file'])) {
            $this->broken_file=$this->config['broken_file'];
            unset($this->config['broken_file']);
        }else{
            $this->broken_file=str_replace('\\','/',dirname(__FILE__)) ."/____mysql_broken_file.php";
        }
        
        $this->broken=$this->get_broken($this->config['broken']);
        unset($this->config['broken']);

//-读写分离
    }

function connect($serverid = 1) {

if(empty($this->config) || empty($this->config[$serverid])) {
            $this->set_broken($serverid);
            $this->halt('config_db_not_found');
        }

$this->link[$serverid] = $this->_dbconnect(
            $this->config[$serverid]['dbhost'],
            $this->config[$serverid]['dbuser'],
            $this->config[$serverid]['dbpw'],
            $this->config[$serverid]['dbcharset'],
            $this->config[$serverid]['dbname'],
            $this->config[$serverid]['pconnect'],
            $serverid
            );
        $this->curlink = $this->link[$serverid];
    }

function _dbconnect($dbhost, $dbuser, $dbpw, $dbcharset, $dbname, $pconnect,$serverid=0) {
        $link = null;
        $func = empty($pconnect) ? 'mysql_connect' : 'mysql_pconnect';
        if(!$link = @$func($dbhost, $dbuser, $dbpw, 1)) {
            $this->set_broken($serverid);
            $this->halt('notconnect');
        } else {
            $this->curlink = $link;
            if($this->version() > '4.1') {
                $dbcharset = $dbcharset ? $dbcharset : $this->config[1]['dbcharset'];
                $serverset = $dbcharset ? 'character_set_connection='.$dbcharset.', character_set_results='.$dbcharset.', character_set_client=binary' : '';
                $serverset .= $this->version() > '5.0.1' ? ((empty($serverset) ? '' : ',').'sql_mode=\'\'') : '';
                $serverset && mysql_query("SET $serverset", $link);
            }
            $dbname && @mysql_select_db($dbname, $link);
        }
        return $link;
    }

function table($tablename) {
        if(!empty($this->map) && !empty($this->map[$tablename])) {
            $id = $this->map[$tablename];
            if(!$this->link[$id]) {
                $this->connect($id);
            }
            $this->curlink = $this->link[$id];
        } else {
            $this->curlink = $this->link[1];
        }
        return $this->tablepre.$tablename;
    }

function select_db($dbname) {
        return mysql_select_db($dbname, $this->curlink);
    }

function fetch_array($query, $result_type = MYSQL_ASSOC) {
        return mysql_fetch_array($query, $result_type);
    }

function fetch_first($sql) {
        return $this->fetch_array($this->query($sql));
    }

function result_first($sql) {
        return $this->result($this->query($sql), 0);
    }

/**
    * 读写分离,并从池中选中随机选中数据库
    * 如果没有read连接将使用write代替
    * 
    * @param string $sql
    */
    function switch_server($sql,$default=1){
        $sql=ltrim($sql);
        if(preg_match('/^select/i',$sql)){
            $type='reads';
        }else{
            $type='writes';
        }

$servers=array_diff($this->$type,$this->filter_broken($this->broken)); 
        if(!$servers&&$type=='reads'){
            $type='writes';
            $servers=array_diff($this->$type,$this->filter_broken($this->broken)); 
        }
        
        $t=count($servers);
        if($t>0){
            //有合理的服务
            /**
            * 根据ip的hash随机抽服务器运行,
            */
            $servers=array_values($servers);
            $onlineip = '';
            if(getenv('HTTP_CLIENT_IP') && strcasecmp(getenv('HTTP_CLIENT_IP'), 'unknown')) {
                $onlineip = getenv('HTTP_CLIENT_IP');
            } elseif(getenv('HTTP_X_FORWARDED_FOR') && strcasecmp(getenv('HTTP_X_FORWARDED_FOR'), 'unknown')) {
                $onlineip = getenv('HTTP_X_FORWARDED_FOR');
            } elseif(getenv('REMOTE_ADDR') && strcasecmp(getenv('REMOTE_ADDR'), 'unknown')) {
                $onlineip = getenv('REMOTE_ADDR');
            } elseif(isset($_SERVER['REMOTE_ADDR']) && $_SERVER['REMOTE_ADDR'] && strcasecmp($_SERVER['REMOTE_ADDR'], 'unknown')) {
                $onlineip = $_SERVER['REMOTE_ADDR'];
            }
            
            !$onlineip&&$onlineip=rand(100);

$i=abs(crc32($onlineip))%$t;

return $servers[$i];
        }
        return $default;
    }
    
    
    function set_broken($server_id){
        $this->broken[$server_id]++;
        if($this->broken_file){
            $str=var_export($this->broken,true);
            $c="<?php 
                \$broken=$str;
            ?>";
            return file_put_contents($this->broken_file,$c);
        }
    }
    
    function filter_broken($broken){
        $error_try=3;
        foreach($broken as $id=>$num){
            if($num<$error_try){
                unset($broken[$id]);
            }
        }
        return array_keys($broken);
    }
    
    function get_broken($realbroken){
        $broken=array();
        $error_try=3;
        if($this->broken_file&&is_file($this->broken_file)){
            @include($this->broken_file);
        }
        settype($realbroken,'array');
        foreach($realbroken as $id){
            $broken[$id]=$error_try;
        }
        return $broken;
    }
    
    /**
    * 读写分离
    * 
    * @param mixed $sql
    * @param mixed $type
    */
    
    function query($sql, $type = '') {

if(defined('DISCUZ_DEBUG') && DISCUZ_DEBUG) {
            $starttime = dmicrotime();
        }
        
        //读写分离
        $server_id=$this->switch_server($sql,1);

if($this->link[$server_id]){
            $this->curlink=$this->link[$server_id];
        }else{
            $this->connect($server_id);
        }
        //-读写分离
        
        
        
        $func = $type == 'UNBUFFERED' && @function_exists('mysql_unbuffered_query') ?
        'mysql_unbuffered_query' : 'mysql_query';
        if(!($query = $func($sql, $this->curlink))) {
            if(in_array($this->errno(), array(2006, 2013)) && substr($type, 0, 5) != 'RETRY') {
                $this->connect($server_id);
                return $this->query($sql, 'RETRY'.$type);
            }
            if($type != 'SILENT' && substr($type, 5) != 'SILENT') {
                $this->halt('query_error', $sql);
            }
        }

if(defined('DISCUZ_DEBUG') && DISCUZ_DEBUG) {
            $this->sqldebug[] = array($sql, number_format((dmicrotime() - $starttime), 6), debug_backtrace());
        }

$this->querynum++;
        return $query;
    }

function affected_rows() {
        return mysql_affected_rows($this->curlink);
    }

function error() {
        return (($this->curlink) ? mysql_error($this->curlink) : mysql_error());
    }

function errno() {
        return intval(($this->curlink) ? mysql_errno($this->curlink) : mysql_errno());
    }

function result($query, $row = 0) {
        $query = @mysql_result($query, $row);
        return $query;
    }

function num_rows($query) {
        $query = mysql_num_rows($query);
        return $query;
    }

function num_fields($query) {
        return mysql_num_fields($query);
    }

function free_result($query) {
        return mysql_free_result($query);
    }

function insert_id() {
        return ($id = mysql_insert_id($this->curlink)) >= 0 ? $id : $this->result($this->query("SELECT last_insert_id()"), 0);
    }

function fetch_row($query) {
        $query = mysql_fetch_row($query);
        return $query;
    }

function fetch_fields($query) {
        return mysql_fetch_field($query);
    }

function version() {
        if(empty($this->version)) {
            $this->version = mysql_get_server_info($this->curlink);
        }
        return $this->version;
    }

function close() {
        return mysql_close($this->curlink);
    }

function halt($message = '', $sql = '') {
        global $_G;
        $dberror = $this->error();
        $dberrno = $this->errno();
        $phperror = '<table style="font-size:11px" cellpadding="0"><tr><td width="270">File</td><td width="80">Line</td><td>Function</td></tr>';
        foreach (debug_backtrace() as $error) {
            $error['file'] = str_replace(DISCUZ_ROOT, '', $error['file']);
            $error['class'] = isset($error['class']) ? $error['class'] : '';
            $error['type'] = isset($error['type']) ? $error['type'] : '';
            $error['function'] = isset($error['function']) ? $error['function'] : '';
            $phperror .= "<tr><td>$error[file]</td><td>$error[line]</td><td>$error[class]$error[type]$error[function]()</td></tr>";
        }
        $phperror .= '</table>';
        $helplink = "http://faq.comsenz.com/?type=mysql&dberrno=".rawurlencode($dberrno)."&dberror=".rawurlencode($dberror);
        @header('Content-Type: text/html; charset='.$_G['config']['output']['charset']);
        echo '<div style="position:absolute;font-size:11px;font-family:verdana,arial;background:#EBEBEB;padding:0.5em;line-height:1.5em">';

echo "<b>PHP Backtrace</b><br />$sql error<br /> $dberror</div>";
        exit();
    }
    
    
    function delete($table, $condition, $limit = 0, $unbuffered = true) {
        if(empty($condition)) {
            $where = '1';
        } elseif(is_array($condition)) {
            $where = $this->implode_field_value($condition, ' AND ');
        } else {
            $where = $condition;
        }
        $sql = "DELETE FROM ".$this->table($table)." WHERE $where ".($limit ? "LIMIT $limit" : '');
        return $this->query($sql, ($unbuffered ? 'UNBUFFERED' : ''));
    }

function insert($table, $data, $return_insert_id = false, $replace = false, $silent = false) {

$sql = $this->implode_field_value($data);

$cmd = $replace ? 'REPLACE INTO' : 'INSERT INTO';

$table = $this->table($table);
        $silent = $silent ? 'SILENT' : '';

$return = $this->query("$cmd $table SET $sql", $silent);

return $return_insert_id ? $this->insert_id() : $return;

}

function update($table, $data, $condition, $unbuffered = false, $low_priority = false) {
        $sql = $this->implode_field_value($data);
        $cmd = "UPDATE ".($low_priority ? 'LOW_PRIORITY' : '');
        $table = $this->table($table);
        $where = '';
        if(empty($condition)) {
            $where = '1';
        } elseif(is_array($condition)) {
            $where = $this->implode_field_value($condition, ' AND ');
        } else {
            $where = $condition;
        }
        $res = $this->query("$cmd $table SET $sql WHERE $where", $unbuffered ? 'UNBUFFERED' : '');
        return $res;
    }

function implode_field_value($array, $glue = ',') {
        $sql = $comma = '';
        foreach ($array as $k => $v) {
            $sql .= $comma."`$k`='$v'";
            $comma = $glue;
        }
        return $sql;
    }

}