基于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; 基本流程为
第一步，先组装查询key，到Cache查询Value，如果存在，继续处理，进入第三步；如果不存在，进入第二步
第二步，根据请求，到DB中，查询相关数据，如果数据存在，把数据放到Cache中
第三步，处理cache或者db中返回的数据
问题
上述流程基本上会出现在每次调用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", ), ） ） 如上所示，通过一个算法，我们可以把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'); } 通过以上方案，我们看到，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"; 在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...)方法的回调 使用以上方法实现对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; } } 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; } 至此，我们使用脚本语言特性，通过user_call_func_array方法补齐所有函数回调机制，从而实现对Cache的封装，通过配置文件定义组装key的规则和每个key的超时时间，再通过Redis的ttl和setnx特性，保证只有一个进程执行DB操作（setnx并非严格意义分布式锁），从而很好避免dogpile问题，实现cache自动触发，保证cache持续存在数据，并且有效减少DB的访问次数，提高性能。
声明：云栖社区站内文章，未经作者本人允许或特别声明，严禁转载，但欢迎分享。
</div>