Memcache分组和多组数据同步机制的实现
【Memcache同步类的实现思想】首先我们明确了解就是Memcache是一个简单、快速、高效的分布式基于内存的缓存工具，一般用于网站等数据库数据缓存、高速交换信息的缓存，比如Session数据等等。Memcache最主要的特点就是两个：一是它是基于内存的高效Hash类缓存方式，二是使用了LUR算法来进行数据有效期控制，这两点就能够很好的满足我们的普通网站的应用。（关于Memcache的使用和协议请参考我的另一篇文章：《Memcache的使用和协议分析详解》http://blog.csdn.net/heiyeshuwu/archive/2006/11/13/1380838.aspx）
Memcache本身也存在一些不足，不能说是不足，至少是为了满足它高效的存取，所以没有去实现的功能，包括：一是本身没有内置分布式功能，无法实现使用多台Memcache服务器来存储不同的数据，最大程度的使用相同的资源；二是无法同步数据，容易造成单点故障。
那么其实都可以通过我们的Memcache的客户端程序来解决的，首先我们PHP的PECL中的Memcache扩展能够有效的解决Memcache的分布式问题，主要的接口就是 addServer() 函数，具体关于addServer()函数的实现可以参考该扩展源代码。那么现在就存在第二个问题，就是说无法同步数据，可以理解为MySQL中Master/Slave的机制，就是说如果我们有多台的Memcache服务器，使用addServer函数的话，每个服务器存储的数据都是唯一的，那么任何一台服务器宕机的话，那么这台服务器上的存储数据将丢失无法访问，这样肯定是无法充分发挥我们Memcache威力的，那么我们就需要一个方式来解决这个问题，这个就是本文主要要探讨解决的问题：关于在多台Memcache服务器中分组和同步数据的问题。
针对这个问题我构建了一个PHP写的Class，大致的思想就是把多台Memcache服务器分成两组，每组的服务器数量可能是一样的（如果总数是复数的话），数量不一样也没关系，写数据的时候往两组服务器都写，数据是一样的，那么A组服务器的数据和B组服务器的数据是一样的，如果A组服务器中某一台机器宕机了，则能够从B组服务器中提取出数据来，这样能够有效的避免单点故障，特别是在高性能要求的网站当中。当然，相应的操作也是有开销的，主要的开销在于网络连接，就是说连接到A组服务器中发现没有数据，则去访问B组服务器。另外还有就是在访问数据过程中，都是从A组开始的，那么并发量很高的情况下，A组服务器的压力比较大，那么我还实现了对A组和B组服务器随机访问的机制，如果在随机的组中没有发现数据则访问另外一个组，这样A组和B组的压力就是1/2了，能够有效的面对高负载的情况。
同样的，这个类是可以扩展的，比如你可以修改成为能够满足两组以上的服务器，但是这样逻辑会更复杂，我基于简便期间，目前就考虑了两组服务器的情况，毕竟一般公司来说，没有很多机器去做Memcache的缓存，所以两组已经能够满足大部分要求了。
PS：如果有兴趣的话，可以完全自己重新实现Memcache的addServer扩展，比如增加addGroup之类的方法，能够更有效的解决同步和分组的问题。同样的，如果想节省资源的使用，可以把Memcached当作线程的方式来运行。
【Memcache同步类的实现代码】
本类经过我大致的各种测试，运行还算稳定，当然，使用时还是请多加注意。
<?php/*** Memcache 操作类 (支持同步和分组访问)* * author : heiyeluren <http://blog.csdn.net/heiyeshuwu>* created : 2007-06-21* lastModifed: 2007-06-21*//*** Memcache操作类*/class MemcacheOpt{//---------------------//属性定义//---------------------/*** 是否进行多组同步 */var $isGroup = true;/*** 多组同步情况下是否使用减轻负载的随机存取 */var $isRandom = true;/*** 默认的Memcache服务器端口 */var $mmcPort = 11211;/*** 保存原始组信息*/var $groups = array();/*** 保存第一、二组Memcache服务器信息*/var $group1 = array();var $group2 = array();/*** 保存第一、二组连接对象*/var $mmc1 = ''; var $mmc2 = ''; //---------------------// 内部操作方法//---------------------/*** 显示错误信息** @param string $msg 需要显示消息的内容* @param string $type 消息类型，error是普通错误消息，fatal 是致命错误，将终止程序执行, message 是普通消息，缺省状态* @return bool true*/function showMessage($msg, $type){$msg .= " ";switch($type){case 'error':echo("Memcache Error: ". $msg);break;case 'fatal':die("Memcache Fatal: ". $msg);break;case 'message':echo("Memcache Message: ". $msg);break;default:echo("Memcache Error: ". $msg);}return true;}/*** 构造函数 (初始化分组和连接到服务器)*/function MemcacheOpt($hostArray, $hostArray2=array()){if (!is_array($hostArray) || empty($hostArray)){$this->showMessage('Memcache host list invalid', 'fatal');}$this->groups = array_merge($hostArray, $hostArray2);$this->splitGroup($hostArray, $hostArray2);$this->connect();}/*** 对组进行切分 (按照是否需要分组进行相应的切分)** @param array $hostArray 主机数组列表1* @param array $hostArray2 主机数组列表2* @return void*/function splitGroup($hostArray, $hostArray2=array()){//如果只有一台机器则不使用分组if (count($hostArray) < 2 && empty($hostArray2)){$this->isGroup = false;}//使用分组if ($this->isGroup){if (is_array($hostArray2) && !empty($hostArray2)){$this->group1 = $hostArray;$this->group2 = $hostArray2;}else{$count = ceil(count($hostArray) / 2);$this->group1 = array_splice($hostArray, 0, $count);$this->group2 = array_splice($hostArray, 0);}}else{$this->group1 = $hostArray; } }/*** 连接到Memcache服务器*/function connect(){if (!is_array($this->group1) || empty($this->group1)){$this->showMessage("Memcache host1 array invalid", 'error');return false;}//连接第一组Memcache服务器$this->mmc1 = new Memcache;foreach($this->group1 as $hosts){$tmp = explode(":", $hosts);$host = $tmp;$port = (!isset($tmp) || $tmp=='') ? $this->mmcPort : $tmp;$this->mmc1->addServer($host, $port);}//如果需要分组则连接第二组Memcache服务器if ($this->isGroup){if ( !is_array($this->group2) || empty($this->group2) ){$this->showMessage("Memcache host2 array invalid", 'error');return false;}$this->mmc2 = new Memcache;foreach($this->group2 as $hosts){$tmp = explode(":", $hosts);$host = $tmp;$port = (!isset($tmp) || $tmp=='') ? $this->mmcPort : $tmp;$this->mmc2->addServer($host, $port);}}}/*** 关闭Memcache服务器连接*/function close(){if (is_object($this->mmc1)){$this->mmc1->close();}if (is_object($this->mmc2)){$this->mmc1->close();} return true;}/*** 数据操作核心函数** @param string $optType 操作类型，主要有 add, set, replace, delete, flush* @param string $key 关键字，如果是 add,set,replace,delete 需要提交key参数* @param string $val 关键字对应的值，如果是 add, set,replace 需要提交value参数* @param int $expire 数据有效期，如果是 add,set,replace需要提交expire参数* @return mixed 不同的需要产生不同的返回*/function opt($optType, $key='', $val='', $expire=''){if (!is_object($this->mmc1)){$this->showMessage("Not availability memcache connection object", 'fatal');}if ($this->isGroup && !is_object($this->mmc2)){$this->showMessage("Group 2 memcache host connection object not availability", 'error');}//加入数据操作if ($optType=='add' || $optType=='set' || $optType=='replace'){$this->mmc1->set($key, $val, false, $expire);if ($this->isGroup && is_object($this->mmc2)){$this->mmc2->set($key, $val, false, $expire);}return true;}//获取数据操作if ($optType == 'get'){//缺省获取第一组数据if (!$this->isGroup || !is_object($this->mmc2)){return $this->mmc1->get($key); }//分组情况下逐组访问$num = ( $this->isRandom ? rand(1, 2) : 1 );$obj = "mmc". $num;$val = $this->$obj->get($key);//如果没有提取到数据，则访问另外一组if ($val == ""){switch($num){case 1: $val = $this->mmc2->get($key); break;case 2: $val = $this->mmc1->get($key); break;default: $val = $this->mmc1->get($key);}}return $val;}//删除数据操作if ($optType == 'delete'){$this->mmc1->delete($key, $expire);if ($this->isGroup && is_object($this->mmc2)){$this->mmc2->delete($key); }return true;}//清空数据操作 if($optType == 'flush'){$this->mmc1->flush();if ($this->isGroup && is_object($this->mmc2)){$this->mmc2->flush(); }return true;}}//---------------------// 外部操作方法//---------------------//增加一个元素function add($key, $val, $expire=''){return $this->opt('add', $key, $val, $expire); }//增加一个元素function set($key, $val, $expire=''){return $this->opt('set', $key, $val, $expire);}//替换一个元素function replace($key, $val, $expire=''){return $this->opt('replace', $val, $expire);}//获取一个元素function get($key){return $this->opt('get', $key);}//删除一个元素function delete($key, $timeout=''){return $this->opt('delete', $key, '', $timeout);}//让所有的元素过期 (本接口不要轻易使用)function flush(){return $this->opt('flush'); }/*** 获取所有Memcache服务器状态*/function getStats(){$status = array();//单独连接到每台Memcacheforeach($this->groups as $key=>$hosts){$tmp = explode(":", $hosts);$host = $tmp;$port = (!isset($tmp) || $tmp=='') ? $this->mmcPort : $tmp;$conn = new Memcache;$conn->connect($host, $port);$s = $conn->getStats();$s['host'] = $host;$s['port'] = $port;$status[$key] = $s;}return $status;}/*** 获取所有Memcache服务器版本号*/function getVersion(){$version = array();$stats = $this->getStats();foreach($stats as $key=>$s){$v['host'] = $s['host'];$v['port'] = $s['port'];$v['version'] = $s['version'];$version[$key] = $v;}return $version;}}?>
【Memcache同步类的测试】
在本机开启多个Memcache守护进程：
[~]$psauxww|grepmemcached
heiyeluren984660.00.260885676??SsJ7:50下午0:00.04/usr/local/memcache/bin/memcached-d-m32-l10.62.240.9-p11214
heiyeluren984370.00.260885676??SsJ7:50下午0:00.04/usr/local/memcache/bin/memcached-d-m32-l10.62.240.9-p11213
heiyeluren984250.00.260885676??SsJ7:50下午0:00.05/usr/local/memcache/bin/memcached-d-m32-l10.62.240.9-p11212
heiyeluren982280.00.260885676??SsJ7:48下午0:00.05/usr/local/memcache/bin/memcached-d-m32-l10.62.240.9-p11211
MemcacheOpt类测试代码：
<?php
require_once("MemcacheOpt.class.php");
//操作代码
$hostArray=array("10.62.240.9","10.62.240.9:11212","10.62.240.9:11213","10.62.240.9:11214");
$m=newMemcacheOpt($hostArray);
$m->add("key1","key1_value",30);
$m->add("key2","key2_value",30);
$m->set("key3","key3_value",30);
$m->set("key4","key4)value",30);
echo$m->get("key1")."";
echo$m->get("key2")."";
echo$m->get("key3")."";
echo$m->get("key4")."";
print_r($m->getStats());
print_r($m->getVersion());
?>
测试代码输出结果：
key1_value
key2_value
key3_value
key4)value
Array
(
[0]=>Array
(
[pid]=>98228
[uptime]=>7440
[time]=>1182433964
[version]=>1.1.12
[rusage_user]=>0.024758
[rusage_system]=>0.034309
[curr_items]=>5
[total_items]=>158
[bytes]=>265
[curr_connections]=>3
[total_connections]=>134
[connection_structures]=>4
[cmd_get]=>87
[cmd_set]=>159
[get_hits]=>81
[get_misses]=>6
[bytes_read]=>6423
[bytes_written]=>27628
[limit_maxbytes]=>33554432
[host]=>10.62.240.9
[port]=>11211
)
[1]=>Array
(
[pid]=>98425
[uptime]=>7359
[time]=>1182433964
[version]=>1.1.12
[rusage_user]=>0.012727
[rusage_system]=>0.038181
[curr_items]=>4
[total_items]=>119
[bytes]=>208
[curr_connections]=>3
[total_connections]=>121
[connection_structures]=>4
[cmd_get]=>45
[cmd_set]=>122
[get_hits]=>45
[get_misses]=>0
[bytes_read]=>4492
[bytes_written]=>22146
[limit_maxbytes]=>33554432
[host]=>10.62.240.9
[port]=>11212
)
[2]=>Array
(
[pid]=>98437
[uptime]=>7356
[time]=>1182433964
[version]=>1.1.12
[rusage_user]=>0.029343
[rusage_system]=>0.022345
[curr_items]=>4
[total_items]=>84
[bytes]=>213
[curr_connections]=>3
[total_connections]=>111
[connection_structures]=>4
[cmd_get]=>88
[cmd_set]=>85
[get_hits]=>43
[get_misses]=>45
[bytes_read]=>4014
[bytes_written]=>23911
[limit_maxbytes]=>33554432
[host]=>10.62.240.9
[port]=>11213
)
[3]=>Array
(
[pid]=>98466
[uptime]=>7330
[time]=>1182433964
[version]=>1.1.12
[rusage_user]=>0.009188
[rusage_system]=>0.035360
[curr_items]=>1
[total_items]=>53
[bytes]=>52
[curr_connections]=>3
[total_connections]=>98
[connection_structures]=>4
[cmd_get]=>37
[cmd_set]=>56
[get_hits]=>28
[get_misses]=>9
[bytes_read]=>2437
[bytes_written]=>21024
[limit_maxbytes]=>33554432
[host]=>10.62.240.9
[port]=>11214
)
)
Array
(
[0]=>Array
(
[host]=>10.62.240.9
[port]=>11211
[version]=>1.1.12
)
[1]=>Array
(
[host]=>10.62.240.9
[port]=>11212
[version]=>1.1.12
)
[2]=>Array
(
[host]=>10.62.240.9
[port]=>11213
[version]=>1.1.12
)
[3]=>Array
(
[host]=>10.62.240.9
[port]=>11214
[version]=>1.1.12
)
)
【结束语】
最后提出两个使用Memcache的小技巧，第一个是如果是一个有效分布式存储的数据，key的取名是很有学问的，这个可以按照项目需要去做，但是key值不要太长，不会冲突就行。第二个就是每个保存在memcache中的数据不要超过1MB。第三个就是开启一个Memcache进程设置内存不要太多也不要太少，按照自己合适设置就行，尽量最大程度提高对硬件的使用，同样可以采取在一台服务器上开启多个memcached来分担一个memcached并发链接的压力。更多技巧在实际使用长慢慢去总结发现，会发现其实memcahe虽然简单，但是很好用。
本文总的就是发现一些简单的东西，可以通过自主扩展的方式来发现更多有趣实用的功能，我想，这就是开源产品的魅力所在吧。欢迎提出批评或者是更好的实现方式。
页:
[1]