基于PHP的一種Cache回調(diào)與自動觸發(fā)技術(shù)

背景

在PHP中使用Memcache或者Redis時，我們一般都會對Memcache和Redis封裝一下，單獨完成寫一個Cache類，作為Memcache活著Redis的代理，且一般為單例模式。在業(yè)務(wù)代碼中，使用Cache類時，操作的基本的示例代碼如下

// cache 的 key
$key = 'this is key';
$expire = 60;// 超時時間

// cache 的實例
$cache = Wk_Cache::instance();
$data = $cache->fetch($key);

// 判斷data
if(empty($data)){
    // 如果為空，調(diào)用db方法
    $db = new Wk_DB();
    $data = $db->getXXX();
    $cache->store($key, $data, $expire);
}
// 處理$data相關(guān)數(shù)據(jù)
return $data;

基本流程為
第一步，先組裝查詢key，到Cache查詢Value，如果存在，繼續(xù)處理，進(jìn)入第三步；如果不存在，進(jìn)入第二步
第二步，根據(jù)請求，到DB中，查詢相關(guān)數(shù)據(jù)，如果數(shù)據(jù)存在，把數(shù)據(jù)放到Cache中
第三步，處理cache或者db中返回的數(shù)據(jù)

問題

上述流程基本上會出現(xiàn)在每次調(diào)用Cache的部分，先cache查詢，沒有的話調(diào)用DB或者第三方接口，獲取數(shù)據(jù)，再次存入Cache，繼續(xù)數(shù)據(jù)處理。多次調(diào)用，既是一種問題，應(yīng)該把這種查詢方式封裝到更底層的方法內(nèi)。而不是每次重復(fù)這樣的邏輯，除了封裝的問題外，還有其他問題，我們統(tǒng)一列舉下

第一：從設(shè)計角度來說 重復(fù)代碼，需要更底層邏輯封裝。
第二：key的組裝，麻煩繁瑣，實際情況，可能會把各種參數(shù)組裝進(jìn)去，維護(hù)的時候，不敢修改。
第三：設(shè)置的expire超時時間，會分散在各處邏輯代碼中，最終很難統(tǒng)計Cache緩存時間的情況。
第四：由于要把cache->store方法放到調(diào)用db之后執(zhí)行，如果db后，還有其他邏輯處理，有可能會忘掉把數(shù)據(jù)放入cache，導(dǎo)致數(shù)據(jù)。
第五：在高并發(fā)系統(tǒng)中，cache失效那一刻，會有大量請求直接穿透到后方，導(dǎo)致DB或者第三方接口壓力陡升，響應(yīng)變慢，進(jìn)一步影響系統(tǒng)穩(wěn)定性，這一現(xiàn)象為“Dogpile”。

以上問題中，最簡單的為2，3種，對于expire超時時間分散的問題，我們可以通過統(tǒng)一配置文件來解決，比如我們可以創(chuàng)建這樣的一個配置文件。

“test"=>array( // namespace，方便分組
             "keys"=> array(
                 “good”=>array(		// 定義的key，此key非最終入cache的key，入key需要和params組裝成唯一的key
                     "timeout"=>600,	// 此處定義超時時間
                     "params"=>array("epid"=>1,"num"=>1),	// 通過這種方法，描述需要傳遞參數(shù)，用于組裝最終入cache的key
                     "desc"=>"描述"
                     ), 
                "top_test"=>array(	// 定義的key，此key非最終入cache的key，入key需要和params組裝成唯一的key
                     "timeout"=>60,	// 此處定義超時時間
                     "ttl"=>10,	// 自動觸發(fā)時間
                     "params"=>array('site_id'=>1,'boutique'=>1,'offset'=>1,'rows'=> 1,'uid'=>1,'tag_id'=>1,'type'=>1),	// 通過這種方法，描述需要傳遞參數(shù)，用于組裝最終入cache的key
                     "desc"=>"描述",
                     "author"=>"ugg",
                     ), 

）
）

如上所示，通過一個算法，我們可以把site_top_feeds和params組裝成唯一的入庫key，組裝后的key，大概是這樣site_top_feeds_site_id=12&boutique=1&offset=0&rows=20&uid=&tag_id=0&type=2通過這種方式，我們避免工人自己組裝key，從而杜絕第二種問題，同時在這個配置文件中，我們也設(shè)置了timeout，這樣調(diào)用store時，我們可以直接從配置文件中讀取，從而避免第三個問題。經(jīng)過如上修改后，我們的cache方法，也做了適當(dāng)?shù)恼{(diào)整，調(diào)用示例如下。

$siteid = 121;
$seminal = 1;
$tag_id = 12;
$tag_id = 22;

$data =  fetch(‘site_top_feeds’,array('site_id'=>$siteid,'boutique'=>$seminal, 'offset'=>"0", 'rows' => "20", 'uid' =>null,’tag_id’=>$tag_id,’type'=>$type),'feed');
if(empty($data)){
// db相關(guān)操作
$db = new Wk_DB();
    $data = $db->getTopFeeds($site_id,$seminal,0,20,null,$tag_id,$type);
//  $data數(shù)據(jù)其他處理邏輯 這里
……

$cache->store(‘site_top_feeds’,$data,array(‘site_id'=>$siteid,'boutique'=>$seminal, 'offset'=>"0", 'rows' => "20", 'uid' =>null,’tag_id’=>$tag_id,’type'=>$type),'feed');
}

通過以上方案，我沒看到，timeout超時時間沒有了，key的組裝也沒有了，對于外層調(diào)用是透明的了。但是我們通過配置文件可以知道site_top_feeds的timeout是多少，通過封裝的算法，知道組裝的key是什么樣的。
這種方式，并沒有解決第一和第四的問題，封裝性；要想完成封裝性，第一件事情要做的就是回調(diào)函數(shù)，作為校本語言，PHP并沒有完善的函數(shù)指針概念，當(dāng)然要想執(zhí)行一個函數(shù)其實也不需要指針。PHP支持回調(diào)函數(shù)的方法有兩種call_user_func，call_user_func_array。
但是，我做了兩個例子，發(fā)現(xiàn)上述方法，執(zhí)行效率比原生方法，相差很多

native:0.0097959041595459s
call_user_func:0.028249025344849s
call_user_func_array:0.046605110168457s

例子代碼如下：

$s = microtime(true);
for($i=0; $i< 10000 ; ++$i){
    $a = new a();
    $data = $a->aaa($array, $array, $array);
    $data = a::bbb($array, $array, $array);
}
$e = microtime(true);
echo "native:".($e-$s)."s\n";

$s = microtime(true);
for($i=0; $i< 10000 ; ++$i){
    $a = new a();
    $data = call_user_func(array($a,'aaa'),$array,$array,$array);
    $data = call_user_func(array('a','bbb'),$array,$array,$array);
}
$e = microtime(true);
echo "call_user_func:".($e-$s)."s\n";

$s = microtime(true);
for($i=0; $i< 10000 ; ++$i){
    $a = new a();
    $data = call_user_func_array(array($a,'aaa'),array(&$array,&$array,&$array));
    $data = call_user_func_array(array('a','bbb'),array(&$array,&$array,&$array));
}
$e = microtime(true);
echo “call_user_func_array:".($e-$s)."s\n";

在PHP中，知道一個對象和方法，其實調(diào)用方法很簡單，比如上面的例子

$a = new a();
$data = $a->aaa($array, $array, $array);
$obj = $a;
$func = ‘a(chǎn)aa’;
$params = array($array,$array,$array);
$obj->$func($params[0],$params[1],$params[2]);	// 通過這種方式可以直接執(zhí)行

這種方式的執(zhí)行性能怎么樣，經(jīng)過我們對比測試發(fā)現(xiàn)

native:0.0092940330505371s
call_user_func:0.028635025024414s
call_user_func_array:0.048038959503174s
my_callback:0.11308288574219s

在加入大量方法策略驗證中，性能損耗比較低，時間消耗僅是原生方法的1.25倍左右，遠(yuǎn)小于call_user_func的3倍多，call_user_func_array的5倍多，具體封裝后的代碼

switch(count($params)){
                case 0: $result = $obj->{$func}();break;
                case 1: $result = $obj->{$func}($params[0]);break;
                case 2: $result = $obj->{$func}($params[0],$params[1]);break;
                case 3: $result = $obj->{$func}($params[0],$params[1],$params[2]);break;
                case 4: $result = $obj->{$func}($params[0],$params[1],$params[2],$params[3]);break;
                case 5: $result = $obj->{$func}($params[0],$params[1],$params[2],$params[3],$params[4]);break;
                case 6: $result = $obj->{$func}($params[0],$params[1],$params[2],$params[3],$params[4],$params[5]);break;
                case 7: $result = $obj->{$func}($params[0],$params[1],$params[2],$params[3],$params[4],$params[5],$params[6]);break;
                default: $result = call_user_func_array(array($obj, $func), $params);  break;
            }   

在使用這種方法之前，考慮過使用create_function來創(chuàng)建匿名函數(shù)，執(zhí)行函數(shù)回調(diào)，經(jīng)過測試create_function只能創(chuàng)造全局函數(shù)，不能創(chuàng)建類函數(shù)和對象函數(shù)，遂放棄。

完成以上準(zhǔn)備工作后，就可以使用回調(diào)機制了，再次調(diào)用的業(yè)務(wù)代碼

….
// 相關(guān)變量賦值
$db = new Wk_DB();
$callback['obj'] = $db;
            $callback['func'] = 'getTopFeeds';
            $callback['params'] = array('site_id'=>$siteid,'boutique'=>$seminal, 'offset'=>"0", 'rows' => "20", 'uid' =>null,'tag_id'=>$tag_id,'type'=>$type);

            $top_feed_list = $cache->smart_fetch('site_top_feeds',$callback,'feed');

使用以上方法實現(xiàn)對cache調(diào)用的封裝，同時保證性能的高效，從而解決第一和第四個問題。

至此已經(jīng)完成前四個問題，從而實現(xiàn)Cache的封裝，并有效的避免了上面提到的第二，第三，第四個問題。但是對于第五個問題，dogpile問題，并沒有解決，針對這種問題，最好的方式是在cache即將失效前，有一個進(jìn)程主動觸發(fā)db操作，獲取DB數(shù)據(jù)放入Cache中，而其他進(jìn)程正常從Cache中獲取數(shù)據(jù)（因為此時cache并未失效）；好在有Redis緩存，我們可以使用Redis的兩個特性很好解決這個問題，先介紹下這兩個接口
TTL方法：以秒為單位，返回給定 key 的剩余生存時間 (TTL, time to live)，當(dāng) key 不存在時，返回 -2 。當(dāng) key 存在但沒有設(shè)置剩余生存時間時，返回 -1 。否則，以秒為單位，返回 key 的剩余生存時間。很明顯，通過這個方法，我們很容易知道key的還剩下的生存時間，通過這個方法，可以在key過期前做點事情，但是光有這個方法還不行，我們需要確保只有進(jìn)程執(zhí)行，而不是所有的進(jìn)程都做，正好用到下面這個方法。
SETNX方法：將 key 的值設(shè)為 value ，當(dāng)且僅當(dāng) key 不存在。若給定的 key 已經(jīng)存在，則SETNX 不做任何動作。SETNX 是『SET if Not eXists』(如果不存在，則 SET) 的簡寫。返回值：設(shè)置成功，返回 1 。設(shè)置失敗，返回 0 。通過這個方法，模擬分布式加鎖，保證只有一個進(jìn)程做執(zhí)行，而其他的進(jìn)程正常處理。結(jié)合以上Redis方法的特性，解決第五種的問題的，實例代碼。

…
// 變量初始化
$key = “this is key”;
$expiration = 600; 
$recalculate_at = 100;
$lock_length = 20;
$data = $cache->fetch($key); 
$ttl = $cache->redis->ttl($key); 
if($recalculate_at>=$ttl&&$r->setnx("lock:".$key,true)){ 
$r->expire(“l(fā)ock:”.$key, $lock_length);
$db = new Wk_DB();
  $data = $db->getXXX();
  $cache->store($key, $expiration, $value);
}

解決方案

好了，關(guān)鍵核心代碼如下
1：function回調(diào)部分代碼

public static function callback($callback){
        // 安全檢查
        if(!isset($callback['obj']) || !isset($callback['func'])
            || !isset($callback['params']) || !is_array($callback['params'])){
            throw new Exception("CallBack Array Error");
        }   
        // 利用反射，判斷對象和函數(shù)是否存在
        $obj = $callback['obj'];
        $func = $callback['func'];
        $params = $callback['params'];
        // 方法判斷        
        $method = new ReflectionMethod($obj,$func);
        if(!$method){
            throw new Exception("CallBack Obj Not Find func");
        }   

        // 方法屬性判斷
        if (!($method->isPublic() || $method->isStatic())) {
            throw new Exception("CallBack Obj func Error");
        }   

        // 參數(shù)個數(shù)判斷(不進(jìn)行逐項檢測)
        $paramsNum = $method->getNumberOfParameters();
        if($paramsNum < count($params)){
            throw new Exception("CallBack Obj Params Error");
        }   

        // 6個參數(shù)以內(nèi)，逐個調(diào)用,超過6個，直接調(diào)用call_user_func_array
        $result = false;
        // 判斷靜態(tài)類方法
        if(!is_object($obj) && $method->isStatic()){
            switch(count($params)){
                case 0: $result = $obj::{$func}();break;
		case 1: $result = $obj::{$func}($params[0]);break;
                case 2: $result = $obj::{$func}($params[0],$params[1]);break;
                case 3: $result = $obj::{$func}($params[0],$params[1],$params[2]);break;
                case 4: $result = $obj::{$func}($params[0],$params[1],$params[2],$params[3]);break;
                case 5: $result = $obj::{$func}($params[0],$params[1],$params[2],$params[3],$params[4]);break;
                case 6: $result = $obj::{$func}($params[0],$params[1],$params[2],$params[3],$params[4],$params[5]);break;
                case 7: $result = $obj::{$func}($params[0],$params[1],$params[2],$params[3],$params[4],$params[5],$params[6]);break;
                default: $result = call_user_func_array(array($obj, $func), $params);  break;
            }
        }else{
            switch(count($params)){
                case 0: $result = $obj->{$func}();break;
                case 1: $result = $obj->{$func}($params[0]);break;
                case 2: $result = $obj->{$func}($params[0],$params[1]);break;
                case 3: $result = $obj->{$func}($params[0],$params[1],$params[2]);break;
                case 4: $result = $obj->{$func}($params[0],$params[1],$params[2],$params[3]);break;
                case 5: $result = $obj->{$func}($params[0],$params[1],$params[2],$params[3],$params[4]);break;
                case 6: $result = $obj->{$func}($params[0],$params[1],$params[2],$params[3],$params[4],$params[5]);break;
                case 7: $result = $obj->{$func}($params[0],$params[1],$params[2],$params[3],$params[4],$params[5],$params[6]);break;
                default: $result = call_user_func_array(array($obj, $func), $params);  break;
            }
        }

2：自動觸發(fā)回調(diào)機制

public function smart_fetch($key,$callback,$namespace="wk") {
	key = $prefix.$key.$suffix;
        $result = $this->_redis->get($key);

        $bttl = false;
        // ttl狀態(tài)判斷(注意冷啟動)
        if(!empty($ttl)){
            // 獲得過期時間
            $rttl = $this->_redis->ttl($key);
            if($rttl > 0 && $ttl >= $rttl &&
                $this->_redis->setnx("lock".$key,true)){
                // 設(shè)置超時時間(超時時間3秒)
                $this->_redis->expire("lock".$key,3);
                $bttl = true;
            }
        }
	// 如何返回值不存在，調(diào)用回調(diào)函數(shù)，獲取數(shù)值，并保持?jǐn)?shù)據(jù)庫
        if($bttl || !$result || (isset($CONFIG['FLUSH']) && !empty($CONFIG['FLUSH']))){
            // 重新調(diào)整參數(shù)
            $callbackparams = array();
            foreach($params as $k=>$value){
                $callbackparams[] = $value;
            }
            $callback['params'] = $callbackparams;
            $result = Wk_Common::callback($callback);
            $expire = $key_config["timeout"];
            // 存儲數(shù)據(jù)
            $status = $this->_redis->setex($key, $expire, $result);
            $result=$this->_redis->get($key);
        }

        // 刪除鎖
        if($bttl){
            $this->_redis->delete("lock".$key);
        }
        return $result;
    }

至此，我們使用腳本語言特性，通過user_call_func_array方法補齊所有函數(shù)回調(diào)機制，從而實現(xiàn)對Cache的封裝，通過配置文件定義組裝key的規(guī)則和每個key的超時時間，再通過Redis的ttl和setnx特性，保證只有一個進(jìn)程執(zhí)行DB操作，從而很好避免dogpile問題，實現(xiàn)cache自動觸發(fā)，保證cache持續(xù)存在數(shù)據(jù)，并且有效減少DB的訪問次數(shù)，提高性能。