kuaileorder/queue/crontab/sync_push_data_test.php

<?php

/**
 * 同步推送数据
 */
include_once(dirname(dirname(dirname(__FILE__)))."/library/publicBase.php");

class syncPushDataTest {
    private $table;

    protected $dbflag = 'pushdatapro';
    private static $db_w = null;
    private static $db_r = null;
    private static $link;

    private $ismaster = false;


    public function __construct() {
        $obj = new mOrder();

        $info = $obj->getSyncDataModifyTime();
        var_dump($info);

        if ($info['pdp_modified'] == '0000-00-00 00:00:00') {
            $sql = "select min(pdp_modified) as pdp_modified from `pdp_tb_trade`";
            $r = $this->execute($sql, false, true);

            $start = $r['pdp_modified'];
        } else {
            $start = $info['pdp_modified'];
        }

        $end = date("Y-m-d H:i:s", strtotime($start)+7200); // 每次取2小时数据

        var_dump($start, $end);


        $list = $this->selectAll('pdp_tb_trade', array('sql'=>'`pdp_modified` >=? and `pdp_modified` <= ?', 'vals'=>array($start, $end)), 'pdp_modified desc');
        var_dump($list);
    }

    /**
     * 解析数据库配置信息
     * 获得数据库的相关连接信息
     * @param sting $dbflag 数据库标识
     * @param boolean $master 是否为主库
     */
    private function parseDbCnf($dbflag,$master=false) {
        $configs = parse_ini_file(_Storage_CNF_PATH, true);
        $config = $configs[strtolower($dbflag)];

        if($master)	{
            $host_port = $config['master'];
        } else {
            $host_port = $config['slave'];
        }
        $hps = explode(',', $host_port);
        $random = rand(0, count($hps)-1);
        $hp = $hps[$random];

        list($host, $port) = explode(':', $hp);

        $cnf = array();
        $cnf['host'] = trim($host);
        $cnf['port'] = trim($port);
        $cnf['user'] = $master==false ? trim($config['user_r']) : trim($config['user']);
        $cnf['pwd'] = $master==false ? trim($config['passwd_r']) : trim($config['passwd']);
        $cnf['db'] = trim($config['db']);
        return $cnf;
    }

    private function getInstance() {
        $dbpara = $this->ismaster==false ? 'db_r' : 'db_w';

        if(!is_null(self::$$dbpara)) {
            self::$link = self::$$dbpara;

            // 判断mysql是否gone away
            $status = self::$link->getAttribute(PDO::ATTR_SERVER_INFO);
            if($status == 'MySQL server has gone away') {
                self::$link = self::$$dbpara = $this->toDb();
            }
        } else {
            self::$link = self::$$dbpara = $this->toDb();
        }
        return self::$link;
    }

    private function initDb($table, $ismaster=false) {
        $this->table = $table;
        $this->ismaster = $ismaster;

        return self::getInstance();
    }

    private function toDb() {
        $type = strtolower($this->dbflag);
        $cnf = $this->parseDbCnf($this->dbflag, $this->ismaster);

        if (count($cnf)<=0 && $cnf['host'] == '') {
            return false;
        }
        try {
            $conn[PDO::ATTR_TIMEOUT] = 3;
            if($GLOBALS['pconnect_db']===true) $conn[PDO::ATTR_PERSISTENT] = true;
            $conn[PDO::MYSQL_ATTR_INIT_COMMAND] = "SET NAMES UTF8;";
            $db = new PDO('mysql:host='.$cnf['host'].';port='.$cnf['port'].';dbname='.$cnf['db'],$cnf['user'],$cnf['pwd'], $conn);
        } catch(PDOException $e) {
            //error_log('['.date('Y-m-d H:i:s').'][first-fail]'.implode('|', $cnf).':'.$e->getMessage()."\n", 3, LOG_PATH_BASE.'/mysql/stat_todb_'.date('Y-m-d').'.log');

            try {
                $conn[PDO::ATTR_TIMEOUT] = 3;
                if($GLOBALS['pconnect_db']===true) $conn[PDO::ATTR_PERSISTENT] = true;
                $conn[PDO::MYSQL_ATTR_INIT_COMMAND] = "SET NAMES UTF8;";
                $db = new PDO('mysql:host='.$cnf['host'].';port='.$cnf['port'].';dbname='.$cnf['db'],$cnf['user'],$cnf['pwd'], $conn);
            } catch(PDOException $e) {
                //error_log('['.date('Y-m-d H:i:s').'][second-fail]'.implode('|', $cnf).':'.$e->getMessage()."\n", 3, LOG_PATH_BASE.'/mysql/stat_todb_'.date('Y-m-d').'.log');
                return false;
            }

            //error_log('['.date('Y-m-d H:i:s').'][second-succ]'.implode('|', $cnf)."\n", 3, LOG_PATH_BASE.'/mysql/stat_todb_'.date('Y-m-d').'.log');
            return $db;
        }

        //error_log('['.date('Y-m-d H:i:s').'][first-succ]'.implode('|', $cnf)."\n", 3, LOG_PATH_BASE.'/mysql/stat_todb_'.date('Y-m-d').'.log');
        return $db;
    }

    /**
     * 数据库Select多数据
     * 封装标准方法
     * @param string $table 表名
     * @param string $where where条件，比如：array('sql'=>'name1=? and name2=?', 'vals'=>array('name1', 'name2'))
     * @param string $orderby 排序，比如："ctime desc, name asc"
     * @param array $limit 读取数量 array(起始数, 数量)
     */
    public function selectAll($table, $where=array(), $orderby='', $limit=array()) {
        $this->initDb($table);

        try {
            $wheresql = '';
            if(!empty($where)) {
                $wheresql = "WHERE {$where['sql']}";
                $vals = $where['vals'];
            }

            $orderby = "ORDER BY {$orderby}";
            if($orderby === null) $orderby = '';

            $limitstr = empty($limit) ? '' : "LIMIT {$limit[0]}, {$limit[1]}";
            $sql = "SELECT * FROM `{$this->table}` {$wheresql} {$orderby} {$limitstr}";
            $st = self::$link->prepare($sql);
            $res = $st->execute($vals);
            if(!$res) return false;

            return $st->fetchAll(PDO::FETCH_ASSOC);

        } catch(PDOException $e) {
            return false;
        }
    }

    /**
     * 执行SQL查询语句
     *
     * 当SQL语句为查询语句时返回执行后的全部数据
     * @access public
     * @param string $sql SQL语句
     * @param boolean $all_rows 是否显示全部数据开关，当为true时，显示全部数据，为false时，显示一行数据，默认为true
     * @param boolean $is_select 是否为查询语句
     * @return array | void
     */
    public function execute($sql, $all_rows=true, $is_select=false) {
        $sql = trim($sql);
        if(!$sql) return false;

        $this->ismaster = $is_select==false ? true : false;
        self::getInstance();
        try {
            $st = self::$link->prepare($sql);
            $res = $st->execute();
            if(!$res) {
                error_log('['.date('Y-m-d H:i:s').']'.$sql.':'.json_encode($st->errorInfo())."\n", 3, LOG_PATH_BASE.'/mysql/execute_'.date('Y-m-d').'.log');
                return false;
            }

            if(!$is_select) return $res;
            if($all_rows==true) {
                return $st->fetchAll(PDO::FETCH_ASSOC);
            } else {
                return $st->fetch(PDO::FETCH_ASSOC);
            }
        } catch(PDOException $e) {
            return false;
        }
    }

}

new syncPushDataTest();