memcached并发CAS模式 memcached

欢迎关注本人公众号

文章图片

应用场景分析：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 = https://www.it610.com/article/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客户端。这个东西还是很强大的，它所使用的libmemcached库是各种语言都支持的。（需要额外安装编译，可以使用PECL装，或手动编译）
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"));

SessionMemd的类如下：
Java代码

文章图片

/**
* 使用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");
?>