基于PHP的一种Cache回调与自动触发技术

1989r 8年前

来自: 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>