Memcache分组和同步机制的实现

[原创] Memcache分组和同步机制的实现

作者：heiyeluren
博客：http://blog.csdn.net/heiyeshuwu
时间：2007-06-25

【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
* 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[0];
            $port = (!isset($tmp[1]) || $tmp[1]=='') ? $this->mmcPort : $tmp[1];
            $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[0];
                $port = (!isset($tmp[1]) || $tmp[1]=='') ? $this->mmcPort : $tmp[1];
                $this->mmc2->addServer($host, $port);
            }
        }
    }

/**
     * 关闭Memcache服务器连接
     */
    function close(){
        if (is_object($this->mmc1)){
            $this->mmc1->close();
        }
        if (is_object($this->mmc1)){
            $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();

//单独连接到每台Memcache
        foreach($this->groups as $key=>$hosts){
            $tmp = explode(":", $hosts);
            $host = $tmp[0];
            $port = (!isset($tmp[1]) || $tmp[1]=='') ? $this->mmcPort : $tmp[1];

$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守护进程：

[~]$ ps auxww | grep memcached
heiyeluren  98466  0.0  0.2  6088 5676  ??  SsJ   7:50下午   0:00.04 /usr/local/memcache/bin/memcached -d -m 32 -l 10.62.240.9 -p 11214
heiyeluren  98437  0.0  0.2  6088 5676  ??  SsJ   7:50下午   0:00.04 /usr/local/memcache/bin/memcached -d -m 32 -l 10.62.240.9 -p 11213
heiyeluren  98425  0.0  0.2  6088 5676  ??  SsJ   7:50下午   0:00.05 /usr/local/memcache/bin/memcached -d -m 32 -l 10.62.240.9 -p 11212
heiyeluren  98228  0.0  0.2  6088 5676  ??  SsJ   7:48下午   0:00.05 /usr/local/memcache/bin/memcached -d -m 32 -l 10.62.240.9 -p 11211

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 = new MemcacheOpt($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虽然简单，但是很好用。

本文总的就是发现一些简单的东西，可以通过自主扩展的方式来发现更多有趣实用的功能，我想，这就是开源产品的魅力所在吧。欢迎提出批评或者是更好的实现方式。

Memcache分组和同步机制的实现相关推荐

FastDFS文件同步机制分析
FastDFS文件同步机制一.tracker server目录及文件结构二.storage server目录及文件结构三.FastDFS文件同步 3.1.同步日志所在目录 3.2.binlog格 ...
内核同步机制-RCU同步机制
转自http://www.360doc.com/content/09/0805/00/36491_4675691.shtml 目录 [隐藏] 1 RCU同步机制 1.1 RCU介绍 1.2 RCU A ...
10、同步机制遵循的原则_我要遵循的10条原则
10.同步机制遵循的原则 by Haseeb Qureshi 由Haseeb Qureshi 我要遵循的10条原则 (10 Principles I Want to Live By) I just c ...
Java多线程的同步机制(synchronized)
一段synchronized的代码被一个线程执行之前,他要先拿到执行这段代码的权限,在 java里边就是拿到某个同步对象的锁(一个对象只有一把锁): 如果这个时候同步对象的锁被其他线程拿走了,他(这个 ...
Nature Neuroscience|群际冲突的脑间同步机制
本文来源:"认知神经科学与学习国家重点实验室"官网编辑:Yezi 审阅:mingzlee7 马燚娜课题组在<Nature Neuroscience> 发表论文揭示群际 ...
Linux 多线程同步机制:互斥量、信号量、条件变量
互斥量:互斥量提供对共享资源的保护访问,它的两种状态:lock和unlock,用来保证某段时间内只有一个线程使用共享资源,互斥量的数据类型是pthread_mutex_t 主要涉及函数:pthread ...
Java高级-线程同步机制实现
2019独角兽企业重金招聘Python工程师标准>>> 前言我们可以在计算机上运行各种计算机软件程序.每一个运行的程序可能包括多个独立运行的线程(Thread). 线程(Threa ...
8天玩转并行开发——第四天同步机制（上）
在并行计算中,不可避免的会碰到多个任务共享变量,实例,集合.虽然task自带了两个方法:task.ContinueWith()和Task.Factory .ContinueWhenAll()来实现任务 ...
转载--线程同步机制及比较
转自:http://blog.csdn.net/eulb/article/details/2177500 先来回答第一个问题,线程实际主要应用于四个主要领域,当然各个领域之间不是绝对孤立的,他们有可能 ...

Memcache分组和同步机制的实现

Memcache分组和同步机制的实现相关推荐

最新文章

热门文章