memcached并发CAS模式

标签：cas sess CAS self SessionMemd 并发 key memcached

http://hudeyong926.iteye.com/blog/1463992

应用场景分析：http://hudeyong926.iteye.com/blog/1172189

如原来MEMCACHED中的KES的内容为A，客户端C1和客户端C2都把A取了出来，C1往准备往其中加B，C2准备往其中加C，这就会造成C1和C2执行后的CACHE KEYS要么是AB要么是AC，而不会出现我们期望的ABC。这种情况，如果不是在集群环境中，而只是单机服务器，可以通过在写CACHE KEYS时增加同步锁，就可以解决问题，可是在集群环境中，同步锁是显然解决不了问题的。

memcached是原子的吗？宏观
所有的被发送到memcached的单个命令是完全原子的。如果您针对同一份数据同时发送了一个set命令和一个get命令，它们不会影响对方。它们将被串行化、先后执行。即使在多线程模式，所有的命令都是原子的；命令序列不是原子的。如果您通过get命令获取了一个item，修改了它，然后想把它set回memcached，我们不保证这个item没有被其他进程（process，未必是操作系统中的进程）操作过。在并发的情况下，您也可能覆写了一个被其他进程set的item。
memcached 1.2.5以及更高版本，提供了gets和cas命令，它们可以解决上面的问题。如果您使用gets命令查询某个key的item，memcached会给您返回该item当前值的唯一标识。如果您覆写了这个item并想把它写回到memcached中，您可以通过cas命令把那个唯一标识一起发送给 memcached。如果该item存放在memcached中的唯一标识与您提供的一致，您的写操作将会成功。如果另一个进程在这期间也修改了这个 item，那么该item存放在memcached中的唯一标识将会改变，您的写操作就会失败。

memcached保存的key value都有一个唯一标识casUnique，在进行incr decr操作时，首先获取casUnique，执行incr，检验返回值是否casUnique+1，如果是，则更新，否则，失败不更新！
尽管这种设计在处理并发时还存在缺陷，但可以通过简单的重试来解决问题！

最初的解决方案：
计划利用memcached的add操作的原子性来控制并发，具体方式如下：
1.申请锁：在校验是否创建过活动前，执行add操作key为memberId，如果add操作失败，则表示有另外的进程在并发的为该memberId创建活动，返回创建失败。否则表示无并发
2.执行创建活动
3.释放锁：创建活动完成后，执行delete操作，删除该memberId。

Java代码



     
   
if (memcache.get(key) == null) {  
// 3 min timeout to avoid mutex holder crash
if (memcache.add(key_mutex, 3 * 60 * 1000) == true) {  
        value = db.get(key);  
        memcache.set(key, value);  
        memcache.delete(key_mutex);  
else
50);  
        retry();  
    }  
}

if (memcache.get(key) == null) {
    // 3 min timeout to avoid mutex holder crash
    if (memcache.add(key_mutex, 3 * 60 * 1000) == true) {
        value = db.get(key);
        memcache.set(key, value);
        memcache.delete(key_mutex);
    } else {
        sleep(50);
        retry();
    }
}

问题：
如此实现存在一些问题：
1.memcached中存放的值有有效期，即过期后自动失效，如add过M1后，M1失效，可以在此add成功
2.即使通过配置，可以使memcached永久有效，即不设有效期，memcached有容量限制，当容量不够后会进行自动替换，即有可能add过M1后，M1被其他key值置换掉，则再次add可以成功。
3.此外，memcached是基于内存的，掉电后数据会全部丢失，导致重启后所有memberId均可重新add。

应对问题：
针对上述的几个问题，根本原因是add操作有时效性，过期，被替换，重启，都会是原来的add操作失效。解决该问题有方法
1.减轻时效性的影响，使用memcached CAS（check and set）方式。

CAS的基本原理
基本原理非常简单，一言以蔽之，就是“版本号”。每个存储的数据对象，多有一个版本号。我们可以从下面的例子来理解：

如果不采用CAS，则有如下的情景：
第一步，A取出数据对象X；
第二步，B取出数据对象X；
第三步，B修改数据对象X，并将其放入缓存；
第四步，A修改数据对象X，并将其放入缓存。
我们可以发现，第四步中会产生数据写入冲突。

如果采用CAS协议，则是如下的情景。
第一步，A取出数据对象X，并获取到CAS-ID1；
第二步，B取出数据对象X，并获取到CAS-ID2；
第三步，B修改数据对象X，在写入缓存前，检查CAS-ID与缓存空间中该数据的CAS-ID是否一致。结果是“一致”，就将修改后的带有CAS-ID2的X写入到缓存。
第四步，A修改数据对象Y，在写入缓存前，检查CAS-ID与缓存空间中该数据的CAS-ID是否一致。结果是“不一致”，则拒绝写入，返回存储失败。

memcached中除了add操作是原子的，还有另外两个操作也是原子的：incr和decr ，使用CAS模式即：
1.预先在memcached中设置一个key值，假设为CREATKEY＝1
2.每次创建活动时，在规则校验前先get出CREATEKEY＝x；
3.进行规则校验
4.执行incr CREATEKEY操作，检验返回值是否为所期望的x＋1，如果不是，则说明在此期间有另外的进程执行了incr操作，即存在并发，放弃更新。否则
5.执行创建活动
这种方法可能存在误判，即本来不存在并发，却被判为并发，如缓存重启，或key值失效后，incr值可能不同于期望值，导致误判。
而使用memcached的CAS方式，可以以几乎0成本的方式解决时效性问题，尽管存在一点小缺陷，但这种缺陷可以通过简单的重试即可解决。考虑实际的产出比，采用memcached的CAS方式更适合实际情况。

具体实现：

1.由于需要使用cas方法，php的memcache客户端不支持该方法，所以改用php的memcached客户端

2. 去除session.save_handler 的设置，然后使用：

Java代码



        
session_set_save_handler(array("SessionMemd","open"),array("SessionMemd","close"),array("SessionMemd","read"),array("SessionMemd","write"),array("SessionMemd","destroy"),array("SessionMemd","gc"));

session_set_save_handler(array("SessionMemd","open"),array("SessionMemd","close"),array("SessionMemd","read"),array("SessionMemd","write"),array("SessionMemd","destroy"),array("SessionMemd","gc"));

SessionMemd的类如下：

Java代码



        
<?php  
/**
* 使用Memcached处理session,纯静态类
* @author yuyii 2010/3/12
*/
class
private static
private static
const RETRY = 3;    //重试次数
public static $maxlife = 3600;  
  
private
  
public static
new
",",$save_path);  
     foreach($servers as &$v) {  
if (preg_match('!tcp://([^:]*):(\d{4,5})!',$v,$match)) {  
1],$match[2]);  
         }  
     }  
     self::$_sess->setOption(Memcached::OPT_DISTRIBUTION, Memcached::DISTRIBUTION_CONSISTENT);  
     self::$_sess->setOption(Memcached::OPT_HASH, Memcached::HASH_CRC);  
return true;  
    }  
  
public static
return true;  
    }  
  
public static
null,$cas);  
     self::$cas = $cas;  
return
    }  
  
public static
if
         $ret = self::$_sess->set($id,$sess_data,self::$maxlife);  
     }  
else
//该方法对于空值会写失败
     }  
  
if ($ret === false) {  
/**
          * 处理验证码失败的情况,让其重试
          */
if ($_SERVER['SCRIPT_NAME'] == "/captcha.php") {  
0;  
while($i++ < self::RETRY) {  
null,$cas);  
if (self::$_sess->cas($cas,$id,$sess_data,self::$maxlife)) return true;  
1000);  
             }  
         }  
return false;  
     }  
else
return true;  
     }  
    }  
  
public static
return
    }  
  
public static
return true;  
    }  
}  
  
//获取设置的生命周期值
SessionMemd::$maxlife = ini_get("session.gc_maxlifetime");    
?>

<?php
/**
* 使用Memcached处理session,纯静态类
* @author yuyii 2010/3/12
*/
class SessionMemd {
  private static $_sess;
  private static $cas;
  const RETRY = 3;    //重试次数
  public static $maxlife = 3600;

  private function __construct() {}

  public static function open($save_path) {
   self::$_sess = new Memcached();
   $servers = explode(",",$save_path);
   foreach($servers as &$v) {
     if (preg_match('!tcp://([^:]*):(\d{4,5})!',$v,$match)) {
       self::$_sess->addServer($match[1],$match[2]);
     }
   }
   self::$_sess->setOption(Memcached::OPT_DISTRIBUTION, Memcached::DISTRIBUTION_CONSISTENT);
   self::$_sess->setOption(Memcached::OPT_HASH, Memcached::HASH_CRC);
   return true;
  }

  public static function close() {
     return true;
  }

  public static function read($id) {
   $ret = self::$_sess->get($id,null,$cas);
   self::$cas = $cas;
   return $ret;
  }

  public static function write($id, $sess_data) {
   if (empty(self::$cas)) {
     $ret = self::$_sess->set($id,$sess_data,self::$maxlife);
   }
   else {
     $ret = self::$_sess->cas(self::$cas,$id,$sess_data,self::$maxlife); //该方法对于空值会写失败
   }

   if ($ret === false) {
     /**
      * 处理验证码失败的情况,让其重试
      */
     if ($_SERVER['SCRIPT_NAME'] == "/captcha.php") {
       $i = 0;
       while($i++ < self::RETRY) {
         $ret = self::$_sess->get($id,null,$cas);
         if (self::$_sess->cas($cas,$id,$sess_data,self::$maxlife)) return true;
         usleep(1000);
       }
     }
     return false;
   }
   else {
     return true;
   }
  }

  public static function destroy($id) {
   return self::$_sess->delete($id);
  }

  public static function gc() {
     return true;
  }
}

//获取设置的生命周期值
SessionMemd::$maxlife = ini_get("session.gc_maxlifetime");  
?>

其中captcha.php 是我这边验证码的文件名，这边可以稍作调整。用回调函数来加载，这样比较灵活。遇到是验证码，程序会去自动重试。

总结：存数据库也会遇到同样的情况，但是数据库锁起来会更方便

标签：cas,sess,CAS,self,SessionMemd,并发,key,memcached
From： https://blog.51cto.com/kenkao/5991671

相关文章

赞助商

阅读排行