从并发处理谈PHP进程间通信（一）外部介质

黄智勇

　　文件是最基本的存储介质，它当然可以作为消息的传输通道来使用。文件的存取各种语言都有各自的多种方案，问题点是多进程并发时的冲突问题。
　　解决存取冲突问题我们使用PHP的 flock() 函数：
　　bool flock ( resource $handle , int $operation [, int &$wouldblock ] )

• $handler 是 使用fopen($path_to_file)获取到的文件句柄；
  
• 　　$operation 是 对文件加锁的方式，有以下值可选：
  　　LOCK_SH (获取共享锁) / LOCK_EX (获取互斥锁) / LOCK_UN (解锁)
  　　这里我们选用互斥锁，一个进程获取到互斥锁后，其他进程在尝试获取锁会被阻塞，直到锁被释放，即实现了自旋；
  　　此外，还有一个参数 LOCK_NB，flock 在获取不到锁时，默认会阻塞住直到锁被其他进程释放，传入 LOCK_NB 与 LOCK_SH 或 LOCK_EX 进行或运算结果（LOCK_EX | LOCK_NB），flock 在锁被其他进程占有时，不会阻塞，而是直接返回 false，这里仅作介绍，我们并不使用它。
  
  
• 　　$wouldblock 参数是一个引用值，在获取不到锁，且不阻塞模式时，$wouldblock 会被设置为 true；（手册中说阻塞时才会被设置为 true。其实我也奇怪这个变量名的。不知道是不是 bug，我的PHP版本是 5.4.5，有知道的烦请解惑）
  
  

代码实现
　　下面是循环ID生成器代码，说明在注释中:
　　

　　
function getCycleIdFromFile($max, $min = 0) {
　　$handler = fopen('/tmp/cycle_id_generator.txt', 'c+');
　　if (!flock($handler, LOCK_EX)) {
　　throw new Exception('error_get_file_lock!');
　　}
　　$cycle_id = trim(fread($handler, 9));
　　$cycle_id++;
　　

　　if ($cycle_id > $max) {
　　$cycle_id = $min;
　　}
　　

　　// 文件指针返回到文件头,并向文件内写入新的cycle_id
　　rewind($handler);
　　fwrite($handler, $cycle_id);
　　

　　// 多写入一些空格为了防止数值升到多位后,突然置为少位后面的数字仍保留
　　fwrite($handler, str_repeat(' ', 9));
　　

　　flock($handler, LOCK_UN);
　　

　　return $cycle_id;
　　
}
　　

mysql
select for update
　　我们常用的 mysql 也可以被当作中间介质来实现进程间的通信，我们规定好某一个数据表内的某一行数据作为消息交换的中转站，使用 mysql 自带的锁来协调多个进程的存取冲突。
　　事务的设计目的就是为了解决多进程并发查询时数据冲突的问题，可是我们常用的事务只能保证数据冲突时会被回滚，数据不会出现错误，并不能实现请求的并行化。对一些数据冲突回滚的请求，需要我们在外层添加逻辑重试。
　　这里介绍 mysql 的一种语法： select for update，会给固定数据加上互斥锁，且另一个请求在获取锁失败时，会阻塞至获取锁成功，mysql 帮我们实现了自旋；
　　用法如下：

• 关闭 mysql 的自动提交，自动提交默认打开，除非使用 transition 语句显示开启事务，默认会将每一条 sql 作为一个事务直接提交执行，这里关闭。 set autocommit=0;
  
• 使用select for update 语句给数据添加互斥锁。注意：需求 mysql 的 innodb 引擎支持;
  
• 进行数据更新和处理操作；
  
• 主动提交事务，并将 自动提交恢复；commit; set autocommit=1;
  

代码实现
　　然后是代码实现：
　　

　　// 数据库连接实现各有不同，demo 可以自己修改一下。
　　function getCycleIdFromMysql($max, $min = 0){
　　Db::db()->execute('set autocommit = 0');

　　$res = Db::db()->qsqlone('SELECT cycle_id FROM cycle_id_generator WHERE>　　

　　$cycle_id = $res['cycle_id'] + 1;
　　if($cycle_id > $max){
　　$cycle_id = $min;
　　}
　　


　　Db::db()->execute("UPDATE cycle_id_generator SET cycle_id = {$cycle_id} WHERE>　　

　　Db::db()->execute('commit');
　　Db::db()->execute('set autocommit = 1');
　　

　　return $cycle_id;
　　}
　　

　　

redis
incr
　　redis 是我们常用的缓存服务器，由于其使用内存存储数据，性能很高。我们使用一个固定的普通键来作为消息中转站，然后利用其 incr 命令的原子性和其执行结果（递增后的值），实现 cycle_id 的递增。
　　incr(key) 若 key 不存在，redis 会先将值设置为0，然后执行递增操作；
　　递增没有问题，可是我们还有个需求是在要其值达到 max 时，再将其置为 min，这时就可能会出现进程A在更新值为 min 时，另一个进程B也检测到值大于了 max，然后将值置为 min，可是这时的值已经不是 max，即发生了值重复更新，那么返回的值必然会有重复；
　　这时，我们就需要自己来实现锁了。
SETNX
　　redis 的 SETNX 命令检测某一个 key 是否存在，若不存在，则将 key 的值设置为 value,并返回结果1； 若 key 已存在，则设置失败，返回值0。
　　SETNX key value
　　它能实现锁是因为它是一个原子命令，即 检测 key 是否存在和设置 key 值在一个事务内，不会出现同时两个进程都检测到 key 不存在，然后同时去设置 key 的情况。
　　我们以另一个值的存在与否，来表示 cycle_id 是否正在被另一个进程修改。
代码实现
　　

　　function getCycleIdFromRedis($max, $min = 0) {
　　$redis = new Redis();
　　$redis->connect('127.0.0.1', 6379);
　　$key_id = 'cycle_id_generator';
　　

　　$cycle_id = $redis->incr($key_id);
　　if ($cycle_id > $max) {
　　// 设置"锁键"的结果 = 获取互斥结果
　　$key_lock = 'cycle_id_lock';
　　if (!$redis->setnx($key_lock, 1)) {
　　return null;
　　}
　　

　　$cycle_id = $min;
　　$redis->set($key_id, $cycle_id);
　　

　　// 最后别忘记释放互斥锁
　　$redis->delete($key_lock);
　　}
　　

　　$redis->close();
　　

　　return $cycle_id;
　　}
　　

　　注意：由于 redis 里没有能实现自旋锁的命令，如果需求最高的获取成功率，我们在检测到 cycle_id 已经是最大值，且试图修改获取锁失败时，退出重试，在外层进行重试。
　　

　　function getCycleId($max, $min = 0) {
　　$cycle_id = getCycleIdFromRedis($max, $min);
　　if (!is_null($cycle_id)) {
　　return $cycle_id;
　　}
　　// 稍微等待下正在更改的进程
　　usleep(500);
　　// 这里使用递归,直至获取成功  并发很高,cycle_id重置很频繁时慎用.
　　return getCycleId($max, $min);
　　}
　　

优化
　　审查代码我们会发现，如果 max-min 的值很小的话，redis 会需要经常重置 key 的值，也就经常需要加锁，重试也就很多。这里，我提供一个优化方法：
　　我们将其 max 设置为一个很大的值（要能被 max-min 整除），返回值时稍做处理，返回 $current % ($max - $min) + $min;。这样，key 需要递增到一个很大的值才会被重置，加锁逻辑和外层逻辑会很少执行到，达到提升效率的目的。
总结：
　　这里简单的评价一下上面所说的三种方法：

• 
  　　性能上没有测试，而且 redis 的性能跟>
  
• 　　代码上非常直观，使用 mysql 非常简洁，而且 redis 要自己实现自旋，比较恶心。
  
  
• 　　实现上，当然是文件最为方便，无任何添加。
  
  

　　本文介绍的都是通过外部介质来进行的通信，下篇介绍下通过 PHP内置函数库来进行进程间通信，欢迎关注；
　　如果您觉得本文对您有帮助，您可以点一下推荐。博客持续更新，欢迎关注。


posted @ 2017-02-13 09:33 枕边书 阅读(...) 评论(...)  编辑 收藏
刷新评论刷新页面返回顶部



公告



Copyright ©2017 枕边书