redis队列_redis实现消息队列

本实例使用thinkphp5

//创建数据库

CREATE TABLE &＃96;think_order_queue&＃96; (
&＃96;id&＃96; int(11) NOT NULL AUTO_INCREMENT,
&＃96;order_id&＃96; int(11) NOT NULL COMMENT &＃39;订单id&＃39;,
&＃96;mobile&＃96; varchar(255) CHARACTER SET utf8mb4 DEFAULT &＃39;0&＃39; COMMENT &＃39;手机号&＃39;,
&＃96;created_at&＃96; int(11) DEFAULT NULL COMMENT &＃39;创建时间&＃39;,
&＃96;updated_at&＃96; int(11) DEFAULT NULL,
&＃96;status&＃96; tinyint(2) DEFAULT NULL COMMENT &＃39;0未处理&＃xff0c;1已处理&＃xff0c;2处理中&＃39;,
PRIMARY KEY (&＃96;id&＃96;)
) ENGINE&＃61;InnoDB DEFAULT CHARSET&＃61;latin1 COMMENT&＃61;&＃39;订单队列表&＃39;;

//创建模型

namespace appcommonmodel;
use thinkModel;
class OrderQueue extends Model{
}

//生成订单

public function orderQueue(){//订单id
$order_id &＃61; rand(10000, 99999);//生成的订单信息
$insert_data &＃61; array(
&＃39;order_id&＃39; &＃61;> $order_id,
&＃39;mobile&＃39; &＃61;> 1857523452 . rand(1, 9),
&＃39;created_at&＃39;&＃61;> time(),
&＃39;status&＃39; &＃61;> 0,
);
$result &＃61; model(&＃39;order_queue&＃39;)->save($insert_data);
if(!$result){
return &＃39;保存失败&＃39;;
}
return "保存成功!";
}

//配送系统处理队列中的订单
public function orderProcess(){
$waiting &＃61; array(&＃39;status&＃39;&＃61;>0); //未处理
$lock &＃61; array(&＃39;status&＃39;&＃61;>2); //改为等待中 //处理数据更新为等待状态
$res_lock &＃61; model(&＃39;order_queue&＃39;)->where($waiting)->order(&＃39;id asc&＃39;)->update($lock);//选择更新的数据进行配送
if($res_lock){
$success &＃61; array(&＃39;status&＃39;&＃61;>1, &＃39;updated_at&＃39;&＃61;>time());
$result &＃61; model(&＃39;order_queue&＃39;)->where($lock)->update($success);
if(!$result){
return &＃39;error&＃39;;
}
return &＃39;success&＃39;;
}
return &＃39;处理失败&＃39;;
}

CREATE TABLE &＃96;redis_queue&＃96; (
&＃96;id&＃96; int(11) NOT NULL AUTO_INCREMENT,
&＃96;uid&＃96; int(11) DEFAULT NULL,
&＃96;time_stamp&＃96; varchar(50) CHARACTER SET utf8mb4 DEFAULT NULL COMMENT &＃39;微秒时间戳&＃39;,
PRIMARY KEY (&＃96;id&＃96;)
) ENGINE&＃61;InnoDB DEFAULT CHARSET&＃61;latin1 COMMENT&＃61;&＃39;redis秒杀&＃39;;

//thinkphp5 thinkcachedirverRedis.php默认没有lpush&＃xff0c;lLen等方法&＃xff0c;要自己添加

public function lLen($key){
return $this->handler->lLen($key);
}
public function lPush($key,$value){
return $this->handler->lPush($key, $value);
}
public function rPush($key,$value){
return $this->handler->rPush($key, $value);
}
public function lpop($key){
return $this->handler->lPop($key);
}
public function close(){
return $this->handler->close();
}

//接受用户请求
public function orderRedis(){//thinkphp5 redis默认没有lpush&＃xff0c;lLen等方法&＃xff0c;要自己添加
$redis &＃61; new cachedriverRedis();
$redis_name &＃61; "miaosha";
for($i&＃61;0; $i<100; $i&＃43;&＃43;){
$uid &＃61; rand(1000, 10000);
$num &＃61; 10;//如果当天人数少于10的时候&＃xff0c;则加入这个队列
if($redis->lLen($redis_name) <10){

//微秒
$redis->rPush($redis_name, $uid . &＃39;%&＃39; . microtime());
echo $uid . &＃39;秒杀成功&＃39;;
} else {//如果当天人数达到10个人&＃xff0c;则返回秒杀已完成
echo &＃39;秒杀已结束&＃39;;
}
}
$redis->close();
}

//处理redis请求
public function redisQueue(){
$redis &＃61; new cachedriverRedis();
$reids_name &＃61; &＃39;miaosha&＃39;;//死循环
while(1){//从队列最左侧取出一个值
$user &＃61; $redis->lpop($reids_name);//然后判断这个值是否存在
if(!$user || $user &＃61;&＃61; &＃39;null&＃39;){sleep(1);
exit();
}//切割事件&＃xff0c;uid
$user_arr &＃61; explode(&＃39;%&＃39;, $user);
$insert_data &＃61; array(
&＃39;uid&＃39; &＃61;> $user_arr[0],
&＃39;time_stamp&＃39; &＃61;> $user_arr[1],
);//保存到数据库中
$res &＃61; model(&＃39;redis_queue&＃39;)->insert($insert_data);//数据库插入失败的时候回滚机制
if(!$res){
$redis->rPush($reids_name, $user);
}sleep(1);
}//释放redis
$redis->close();
}