基于PHP的一種Cache回調與自動觸發技術
來自： https://yq.aliyun.com/articles/6040
背景
在PHP中使用Memcache或者Redis時，我們一般都會對Memcache和Redis封裝一下，單獨完成寫一個Cache類，作為Memcache或者Redis的代理，且一般為單例模式。在業務代碼中，使用Cache類時，操作的基本的示例代碼如下
// cache 的 key $key = 'this is key'; $expire = 60;// 超時時間// cache 的實例 $cache = Wk_Cache::instance(); $data = $cache->fetch($key);
// 判斷data if(empty($data)){ // 如果為空，調用db方法 $db = new Wk_DB(); $data = $db->getXXX(); $cache->store($key, $data, $expire); } // 處理$data相關數據 return $data;</pre>
基本流程為
第一步，先組裝查詢key，到Cache查詢Value，如果存在，繼續處理，進入第三步；如果不存在，進入第二步
第二步，根據請求，到DB中，查詢相關數據，如果數據存在，把數據放到Cache中
第三步，處理cache或者db中返回的數據
</div>
問題
上述流程基本上會出現在每次調用Cache的部分，先cache查詢，沒有的話調用DB或者第三方接口，獲取數據，再次存入Cache，繼續數據處理。單獨看這樣的代碼，邏輯合理，但是如果所有調用緩存的地方都是這樣的方式多就是一種問題了，應該把這種查詢方式封裝到更底層的方法內，而不是每次重復這樣的邏輯，除了封裝的問題外，還有其他一些問題，我們統一列舉下
第一：從設計角度來說 重復代碼，需要更底層邏輯封裝。
第二：key的組裝，麻煩繁瑣，實際情況，可能會把各種參數組裝進去，維護的時候，不敢輕易修改。
第三：設置的expire超時時間，會分散在各處邏輯代碼中，最終很難統計Cache緩存時間的情況。
第四：由于要把cache->store方法放到調用db之后執行，如果db后，還有其他邏輯處理，有可能會忘掉把數據放入cache，存在調試風險。
第五：最重要的是高并發系統中，cache失效那一刻，會有大量請求直接穿透到后方，導致DB或者第三方接口壓力陡升，響應變慢，進一步影響系統穩定性，這一現象為“Dogpile”。
以上問題中，最簡單的是2，3問題。對于expire超時時間分散的問題，我們可以通過統一配置文件來解決，比如我們可以創建這樣的一個配置文件。
“test"=>array( // namespace，方便分組 "keys"=> array( “good”=>array( // 定義的key，此key非最終入cache的key，入key需要和params組裝成唯一的key "timeout"=>600, // 此處定義超時時間 "params"=>array("epid"=>1,"num"=>1), // 通過這種方法，描述需要傳遞參數，用于組裝最終入cache的key "desc"=>"描述" ), "top_test"=>array( // 定義的key，此key非最終入cache的key，入key需要和params組裝成唯一的key "timeout"=>60, // 此處定義超時時間 "ttl"=>10, // 自動觸發時間 "params"=>array('site_id'=>1,'boutique'=>1,'offset'=>1,'rows'=> 1,'uid'=>1,'tag_id'=>1,'type'=>1), // 通過這種方法，描述需要傳遞參數，用于組裝最終入cache的key "desc"=>"描述", "author"=>"ugg", ),） ）</pre>
如上所示，通過一個算法，我們可以把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，從而杜絕第二種問題，在這個配置文件中，我們也設置了timeout，這樣調用store時，我們可以直接從配置文件中讀取，從而避免第三個問題。經過如上修改后，我們的cache方法，也做了適當的調整，調用示例如下。
$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相關操作 $db = new Wk_DB(); $data = $db->getTopFeeds($site_id,$seminal,0,20,null,$tag_id,$type); // $data數據其他處理邏輯 這里 ……
$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'); }</pre>
通過以上方案，我們看到，timeout超時時間沒有了，key的組裝也沒有了，對于外層調用是透明的了。我們通過配置文件可以知道site_top_feeds的timeout是多少，通過封裝的算法，知道組裝的key是什么樣的。
這種方式，并沒有解決第一和第四的問題，封裝性；要想完成封裝性，第一件事情要做的就是回調函數，PHP作為腳本語言，并沒有完善的函數指針概念，當然要想執行一個函數其實也不需要指針。PHP支持回調函數的方法有兩種call_user_func，call_user_func_array。但是，經過測試會發現上述方法，執行效率比原生方法差很多
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";</pre>
在PHP中，知道一個對象和方法，其實調用方法很簡單，比如上面的例子
$a = new a(); $data = $a->aaa($array, $array, $array); $obj = $a; $func = ‘aaa’; $params = array($array,$array,$array); $obj->$func($params[0],$params[1],$params[2]); // 通過這種方式可以直接執行詳細代碼：
$s = microtime(true); for($i=0; $i< 10000 ; ++$i){ $obj = new a(); $func = 'aaa'; $params = array($array,$array,$array); $obj->$func($params[0],$params[1],$params[2]); // 通過這種方式可以直接執行 } $e = microtime(true); echo "my_callback:".($e-$s)."s\n";這種方式的執行性能怎么樣，經過我們對比測試發現
native:0.0092940330505371s call_user_func:0.028635025024414s call_user_func_array:0.048038959503174s my_callback:0.011308288574219s在加入大量方法策略驗證中，性能損耗比較低，時間消耗僅是原生方法的1.25倍左右，遠小于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;// 超過7項數據后變態方法，采用call_user_func_array機制，作為保底方案 }備注：在使用這種方法之前，考慮過使用create_function來創建匿名函數，執行函數回調，經過測試create_function只能創造全局函數，不能創建類函數和對象函數，遂放棄。
完成以上準備工作后，就可以使用回調機制了，再次調用的業務代碼
…. // 相關變量賦值 $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');// smart_fetch第一步會先從cache取數據，如果cache無數據，會自動觸發$db->getTopFeeds($site_id...)方法的回調</pre>使用以上方法實現對cache調用的封裝，同時保證性能的高效，從而解決第一和第四個問題。
至此已經完成前四個問題，從而實現Cache的封裝，并有效的避免了上面提到的第二，第三，第四個問題。但是對于第五個問題，dogpile問題，并沒有解決，針對這種問題，最好的方式是在cache即將失效前，有一個進程主動觸發DB操作，獲取DB數據放入Cache中，而其他進程正常從Cache中獲取數據（因為此時Cache并未失效）；好在有Redis緩存可以選擇，我們可以使用Redis的兩個特性很好解決這個問題，先介紹下這兩個接口
TTL方法：以秒為單位，返回給定 key 的剩余生存時間 (TTL, time to live)，當 key 不存在時，返回 -2 。當 key 存在但沒有設置剩余生存時間時，返回 -1 。否則，以秒為單位，返回 key 的剩余生存時間。很明顯，通過這個方法，我們很容易知道key的還剩下的生存時間，通過這個方法，可以在key過期前做點事情，但是光有這個方法還不行，我們需要確保只有一個進程執行，而不是所有的進程都做，正好用到下面這個方法。
SETNX方法：將 key 的值設為 value ，當且僅當 key 不存在。若給定的 key 已經存在，則SETNX 不做任何動作。SETNX 是『SET if Not eXists』(如果不存在，則 SET) 的簡寫。返回值：設置成功，返回 1 。設置失敗，返回 0 。通過這個方法，模擬分布式加鎖，保證只有一個進程做執行，而其他的進程正常處理。結合以上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(“lock:”.$key, $lock_length); $db = new Wk_DB(); $data = $db->getXXX(); $cache->store($key, $expiration, $value); }解決方案
好了，關鍵核心代碼如下
1：function回調部分代碼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"); }
// 利用反射，判斷對象和函數是否存在 $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"); } // 參數個數判斷(不進行逐項檢測) $paramsNum = $method->getNumberOfParameters(); if($paramsNum < count($params)){ throw new Exception("CallBack Obj Params Error"); } // 6個參數以內，逐個調用,超過6個，直接調用call_user_func_array $result = false; // 判斷靜態類方法 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; } }</pre>2：自動觸發回調機制
public function smart_fetch($key,$callback,$namespace="wk") { key = $prefix.$key.$suffix; $result = $this->_redis->get($key);$bttl = false; // ttl狀態判斷(注意冷啟動) if(!empty($ttl)){ // 獲得過期時間 $rttl = $this->_redis->ttl($key); if($rttl > 0 && $ttl >= $rttl && $this->_redis->setnx("lock".$key,true)){ // 設置超時時間(超時時間3秒) $this->_redis->expire("lock".$key,3); $bttl = true; } } // 如何返回值不存在，調用回調函數，獲取數值，并保持數據庫 if($bttl || !$result || (isset($CONFIG['FLUSH']) && !empty($CONFIG['FLUSH']))){ // 重新調整參數 $callbackparams = array(); foreach($params as $k=>$value){ $callbackparams[] = $value; } $callback['params'] = $callbackparams; $result = Wk_Common::callback($callback); $expire = $key_config["timeout"]; // 存儲數據 $status = $this->_redis->setex($key, $expire, $result); $result=$this->_redis->get($key); } // 刪除鎖 if($bttl){ $this->_redis->delete("lock".$key); } return $result; }</pre>至此，我們使用腳本語言特性，通過user_call_func_array方法補齊所有函數回調機制，從而實現對Cache的封裝，通過配置文件定義組裝key的規則和每個key的超時時間，再通過Redis的ttl和setnx特性，保證只有一個進程執行DB操作（setnx并非嚴格意義分布式鎖），從而很好避免dogpile問題，實現cache自動觸發，保證cache持續存在數據，并且有效減少DB的訪問次數，提高性能。
聲明：云棲社區站內文章，未經作者本人允許或特別聲明，嚴禁轉載，但歡迎分享。
</div>本文由用戶 1989r 自行上傳分享，僅供網友學習交流。所有權歸原作者，若您的權利被侵害，請聯系管理員。轉載本站原創文章，請注明出處，并保留原始鏈接、圖片水印。本站是一個以用戶分享為主的開源技術平臺，歡迎各類分享！