PHP解决抢购、秒杀、抢楼、抽奖等阻塞式高并发库存防控超量的思路方法

PHP解决抢购、秒杀、抢楼、抽奖等阻塞式高并发库存防控超量的思路方法

如今在电商行业里,秒杀抢购活动已经是商家常用促销手段。但是库存数量有限，而同时下单人数超过了库存量，就会导致商品超卖甚至库存变负数的问题。
又比如：抢购火车票、论坛抢楼、抽奖乃至爆红微博评论等也会引发阻塞式高并发问题。如果不做任何措施可能在高瞬间造成服务器瘫痪，如何解决这个问题呢？

解决这个问题比较流行的思路：

1.用额外的单进程处理一个队列，下单请求放到队列里，一个个处理，就不会有并发的问题了，但是要额外的后台进程以及延迟问题，不予考虑。2.数据库乐观锁，大致的意思是先查询库存，然后立马将库存+1，然后订单生成后，在更新库存前再查询一次库存，看看跟预期的库存数量是否保持一致，不一致就回滚，提示用户库存不足。3.根据update结果来判断，我们可以在sql2的时候加一个判断条件update ... where 库存>0，如果返回false，则说明库存不足，并回滚事务。4.借助文件排他锁，在处理下单请求的时候，用flock锁定一个文件，如果锁定失败说明有其他订单正在处理，此时要么等待要么直接提示用户"服务器繁忙"

注意：

将库存字段number字段设为unsigned，当库存为0时，因为字段不能为负数，将会返回false

//库存减少  
$sql="update ih_store set number=number-{$number} where sku_id='$sku_id' and number>0";  
$store_rs=mysql_query($sql,$conn);    
if(mysql_affected_rows()){    
    insertLog('库存减少成功');  
}  

关于悲观锁和乐观锁：

悲观锁(Pessimistic Lock), 顾名思义，就是很悲观，每次去拿数据的时候都认为别人会修改，所以每次在拿数据的时候都会上锁，这样别人想拿这个数据就会block直到它拿到锁。传统的关系型数据库里边就用到了很多这种锁机制，比如行锁，表锁等，读锁，写锁等，都是在做操作之前先上锁。

乐观锁(Optimistic Lock), 顾名思义，就是很乐观，每次去拿数据的时候都认为别人不会修改，所以不会上锁，但是在更新的时候会判断一下在此期间别人有没有去更新这个数据，可以使用版本号等机制。乐观锁适用于多读的应用类型，这样可以提高吞吐量，像数据库如果提供类似于write_condition机制的其实都是提供的乐观锁。

两种锁各有优缺点，不可认为一种好于另一种，像乐观锁适用于写比较少的情况下，即冲突真的很少发生的时候，这样可以省去了锁的开销，加大了系统的整个吞吐量。但如果经常产生冲突，上层应用会不断的进行retry，这样反倒是降低了性能，所以这种情况下用悲观锁就比较合适。

方案一：使用消息队列来实现

使用redis队列，因为pop操作是原子的，即使有很多用户同时到达，也是依次执行，推荐使用
可以基于例如MemcacheQ等这样的消息队列，具体的实现方案这么表述吧
比如有100张票可供用户抢，那么就可以把这100张票放到缓存中，读写时不要加锁。 当并发量大的时候，可能有500人左右抢票成功，这样对于500后面的请求可以直接转到活动结束的静态页面。进去的500个人中有400个人是不可能获得商品的。所以可以根据进入队列的先后顺序只能前100个人购买成功。后面400个人就直接转到活动结束页面。当然进去500个人只是举个例子，至于多少可以自己调整。而活动结束页面一定要用静态页面，不要用数据库。这样就减轻了数据库的压力。

先将商品库存如队列

<?php  
$store=1000;  
$redis=new Redis();  
$result=$redis->connect('127.0.0.1',6379);  
$res=$redis->llen('goods_store');  
echo $res;  
$count=$store-$res;  
for($i=0;$i<$count;$i++){  
    $redis->lpush('goods_store',1);  //每次往列表末尾放入一个元素
}  
echo $redis->llen('goods_store');  
?>  
抢购、描述逻辑

18deb

<?php  
$conn=mysql_connect("localhost","big","123456");    
if(!$conn){    
    echo "connect failed";    
    exit;    
}   
mysql_select_db("big",$conn);   
mysql_query("set names utf8");  
  
$price=10;  
$user_id=1;  
$goods_id=1;  
$sku_id=11;  
$number=1;  
  
//生成唯一订单号  
function build_order_no(){  
    return date('ymd').substr(implode(NULL, array_map('ord', str_split(substr(uniqid(), 7, 13), 1))), 0, 8);  
}  
//记录日志  
function insertLog($event,$type=0){  
    global $conn;  
    $sql="insert into ih_log(event,type)   
    values('$event','$type')";    
    mysql_query($sql,$conn);    
}  
  
//模拟下单操作  
//下单前判断redis队列库存量  
$redis=new Redis();  
$result=$redis->connect('127.0.0.1',6379);  
$count=$redis->lpop('goods_store');  
if(!$count){  
    insertLog('error:no store redis');  
    return;  
}  
  
//生成订单    
$order_sn=build_order_no();  
$sql="insert into ih_order(order_sn,user_id,goods_id,sku_id,price)   
values('$order_sn','$user_id','$goods_id','$sku_id','$price')";    
$order_rs=mysql_query($sql,$conn);   
  
//库存减少  
$sql="update ih_store set number=number-{$number} where sku_id='$sku_id'";  
$store_rs=mysql_query($sql,$conn);    
if(mysql_affected_rows()){    
    insertLog('库存减少成功');  
}else{    
    insertLog('库存减少失败');  
}   

[b]方案二：当有多台服务器时，可以采用分流的形式实现[/b]
假设有m张票, 有n台产品服务器接收请求,有x个请求路由服务器随机转发
直接给每台产品服务器分配 m/n张票
每台产品服务器内存做计数器,比如允许m/n*(1+0.1)个人进来。
当内存计数器已满：
后面进的人, 直接跳到到转到活动结束的静态页面，
通知路由服务器,不在路由到这台服务器(这个值得商讨)。
所有产品服务器进来的m/n*(1+0.1)个人再全部转发到一台付款服务器上,进入付款环节,看谁手快了,这时候人少,加锁什么的就简单的。

方案三、如果是单服务器，可以使用Memcache锁来实现

使用mysql的事务，锁住操作的行

<?php  
$conn=mysql_connect("localhost","big","123456");    
if(!$conn){    
    echo "connect failed";    
    exit;    
}   
mysql_select_db("big",$conn);   
mysql_query("set names utf8");  
  
$price=10;  
$user_id=1;  
$goods_id=1;  
$sku_id=11;  
$number=1;  
  
//生成唯一订单号  
function build_order_no(){  
    return date('ymd').substr(implode(NULL, array_map('ord', str_split(substr(uniqid(), 7, 13), 1))), 0, 8);  
}  
//记录日志  
function insertLog($event,$type=0){  
    global $conn;  
    $sql="insert into ih_log(event,type)   
    values('$event','$type')";    
    mysql_query($sql,$conn);    
}  
  
//模拟下单操作  
//库存是否大于0  
mysql_query("BEGIN");   //开始事务  
$sql="select number from ih_store where goods_id='$goods_id' and sku_id='$sku_id' FOR UPDATE";//此时这条记录被锁住,其它事务必须等待此次事务提交后才能执行  
$rs=mysql_query($sql,$conn);  
$row=mysql_fetch_assoc($rs);  
if($row['number']>0){  
    //生成订单   
    $order_sn=build_order_no();   
    $sql="insert into ih_order(order_sn,user_id,goods_id,sku_id,price)   
    values('$order_sn','$user_id','$goods_id','$sku_id','$price')";    
    $order_rs=mysql_query($sql,$conn);   
      
    //库存减少  
    $sql="update ih_store set number=number-{$number} where sku_id='$sku_id'";  
    $store_rs=mysql_query($sql,$conn);    
    if(mysql_affected_rows()){    
        insertLog('库存减少成功');  
        mysql_query("COMMIT");//事务提交即解锁  
    }else{    
        insertLog('库存减少失败');  
    }  
}else{  
    insertLog('库存不够');  
    mysql_query("ROLLBACK");  
}  
?>  

product_key 为票的key
product_lock_key 为票锁key
当product_key存在于memcached中时，所有用户都可以进入下单流程。
当进入支付流程时，首先往memcached存放add(product_lock_key, “1″),
如果返回成功，进入支付流程。
如果不成，则说明已经有人进入支付流程，则线程等待N秒，递归执行add操作。

方案四、借助文件排他锁

在处理下单请求的时候，用flock锁定一个文件，如果锁定失败说明有其他订单正在处理，此时要么等待要么直接提示用户"服务器繁忙"
本文要说的是第4种方案，大致代码如下

<?php  
$conn=mysql_connect("localhost","root","123456");    
if(!$conn){    
    echo "connect failed";    
    exit;    
}   
mysql_select_db("big-bak",$conn);   
mysql_query("set names utf8");  
  
$price=10;  
$user_id=1;  
$goods_id=1;  
$sku_id=11;  
$number=1;  
  
//生成唯一订单号  
function build_order_no(){  
    return date('ymd').substr(implode(NULL, array_map('ord', str_split(substr(uniqid(), 7, 13), 1))), 0, 8);  
}  
//记录日志  
function insertLog($event,$type=0){  
    global $conn;  
    $sql="insert into ih_log(event,type)   
    values('$event','$type')";    
    mysql_query($sql,$conn);    
}  
  
$fp = fopen("lock.txt", "w+");  
if(!flock($fp,LOCK_EX | LOCK_NB)){  
    echo "系统繁忙，请稍后再试";  
    return;  
}  
//下单  
$sql="select number from ih_store where goods_id='$goods_id' and sku_id='$sku_id'";  
$rs=mysql_query($sql,$conn);  
$row=mysql_fetch_assoc($rs);  
if($row['number']>0){//库存是否大于0  
    //模拟下单操作   
    $order_sn=build_order_no();   
    $sql="insert into ih_order(order_sn,user_id,goods_id,sku_id,price)   
    values('$order_sn','$user_id','$goods_id','$sku_id','$price')";    
    $order_rs=mysql_query($sql,$conn);   
      
    //库存减少  
    $sql="update ih_store set number=number-{$number} where sku_id='$sku_id'";  
    $store_rs=mysql_query($sql,$conn);    
    if(mysql_affected_rows()){    
        insertLog('库存减少成功');  
        flock($fp,LOCK_UN);//释放锁  
    }else{    
        insertLog('库存减少失败');  
    }   
}else{  
    insertLog('库存不够');  
}  
fclose($fp);  

阻塞(等待)模式：

<?php
$fp = fopen("lock.txt", "w+");
if(flock($fp,LOCK_EX))
{
//..处理订单
flock($fp,LOCK_UN);
}
fclose($fp);
?>

非阻塞模式：

<?php
$fp = fopen("lock.txt", "w+");
if(flock($fp,LOCK_EX | LOCK_NB))
{
//..处理订单
flock($fp,LOCK_UN);
}
else
{
echo "系统繁忙，请稍后再试";
}

fclose($fp);
?>

[b]模拟5000高并发测试[/b]webbench -c 5000 -t 60 http://192.168.1.198/big/index.php ab -r -n 6000 -c 5000  http://192.168.1.198/big/index.php（url）

测试数据表

[php] view plain copy--  
-- 数据库: `big`  
--  
  
-- --------------------------------------------------------  
  
--  
-- 表的结构 `ih_goods`  
--  
  
  
CREATE TABLE IF NOT EXISTS `ih_goods` (  
  `goods_id` int(10) unsigned NOT NULL AUTO_INCREMENT,  
  `cat_id` int(11) NOT NULL,  
  `goods_name` varchar(255) NOT NULL,  
  PRIMARY KEY (`goods_id`)  
) ENGINE=MyISAM  DEFAULT CHARSET=utf8 AUTO_INCREMENT=2 ;  
  
  
--  
-- 转存表中的数据 `ih_goods`  
--  
  
  
INSERT INTO `ih_goods` (`goods_id`, `cat_id`, `goods_name`) VALUES  
(1, 0, '小米手机');  
  
-- --------------------------------------------------------  
  
--  
-- 表的结构 `ih_log`  
--  
  
CREATE TABLE IF NOT EXISTS `ih_log` (  
  `id` int(11) NOT NULL AUTO_INCREMENT,  
  `event` varchar(255) NOT NULL,  
  `type` tinyint(4) NOT NULL DEFAULT '0',  
  `addtime` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,  
  PRIMARY KEY (`id`)  
) ENGINE=MyISAM DEFAULT CHARSET=utf8 AUTO_INCREMENT=1 ;  
  
--  
-- 转存表中的数据 `ih_log`  
--  
  
  
-- --------------------------------------------------------  
  
--  
-- 表的结构 `ih_order`  
--  
  
CREATE TABLE IF NOT EXISTS `ih_order` (  
  `id` int(11) NOT NULL AUTO_INCREMENT,  
  `order_sn` char(32) NOT NULL,  
  `user_id` int(11) NOT NULL,  
  `status` int(11) NOT NULL DEFAULT '0',  
  `goods_id` int(11) NOT NULL DEFAULT '0',  
  `sku_id` int(11) NOT NULL DEFAULT '0',  
  `price` float NOT NULL,  
  `addtime` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP,  
  PRIMARY KEY (`id`)  
) ENGINE=InnoDB DEFAULT CHARSET=utf8 COMMENT='订单表' AUTO_INCREMENT=1 ;  
  
--  
-- 转存表中的数据 `ih_order`  
--  
  
  
-- --------------------------------------------------------  
  
--  
-- 表的结构 `ih_store`  
--  
  
CREATE TABLE IF NOT EXISTS `ih_store` (  
  `id` int(11) NOT NULL AUTO_INCREMENT,  
  `goods_id` int(11) NOT NULL,  
  `sku_id` int(10) unsigned NOT NULL DEFAULT '0',  
  `number` int(10) NOT NULL DEFAULT '0',  
  `freez` int(11) NOT NULL DEFAULT '0' COMMENT '虚拟库存',  
  PRIMARY KEY (`id`)  
) ENGINE=InnoDB  DEFAULT CHARSET=utf8 COMMENT='库存' AUTO_INCREMENT=2 ;  
  
--  
-- 转存表中的数据 `ih_store`  
--  
  
INSERT INTO `ih_store` (`id`, `goods_id`, `sku_id`, `number`, `freez`) VALUES  
(1, 1, 11, 500, 0);