PHP(Mysql/Redis)消息队列的介绍及应用场景案例--转载

CREATE TABLE `order_queue` (  
  `id` int(11) unsigned NOT NULL AUTO_INCREMENT COMMENT ‘订单的id号‘,  
  `order_id` int(11) NOT NULL,  
  `mobile` varchar(20) NOT NULL COMMENT ‘用户的手机号‘,  
  `address` varchar(100) NOT NULL COMMENT ‘用户的地址‘,  
  `created_at` datetime NOT NULL DEFAULT ‘0000-00-00 00:00:00‘ COMMENT ‘订单创建的时间‘,  
  `updated_at` datetime NOT NULL DEFAULT ‘0000-00-00 00:00:00‘ COMMENT ‘物流系统处理完成的时间‘,  
  `status` tinyint(2) NOT NULL COMMENT ‘当前状态，0 未处理，1 已处理，2处理中‘,  
  PRIMARY KEY (`id`)  
) ENGINE=InnoDB  DEFAULT CHARSET=utf8;  

<?php   
include ‘class/db.php‘;  
  
if(!empty($_GET[‘mobile‘])){  
    $order_id = rand(10000,99999);  
    $insert_data = array(  
        ‘order_id‘=>$order_id,  
        ‘mobile‘=>$_GET[‘mobile‘],      //记得过滤
        ‘created_at‘=>date(‘Y-m-d H:i:s‘,time()),  
        ‘order_id‘=>$order_id,  
        ‘status‘=>0,    //0，未处理状态
    );  
    $db = DB::getIntance();  
//把数据放入队列表中
    $res = $db->insert(‘order_queue‘,$insert_data);  
    if($res){  
        echo $insert_data[‘order_id‘]."保存成功";  
    }else{  
        echo "保存失败";  
    }  
}else{  
    echo "1";  
}  
?> 

<?php   
//配送系统处理订单并进行标记
include ‘class/db.php‘;  
$db = DB::getIntance();  
//1：先要把要处理的数据状态改为待处理
$waiting = array(‘status‘=>0,);  
$lock = array(‘status‘=>2,);  
$res_lock = $db->update(‘order_queue‘,$lock,$waiting,2);  
//2：选择出刚刚更新的数据，然后进行配送系统处理
if($res_lock){  
    //选择出要处理的订单内容
    $res = $db->selectAll(‘order_queue‘,$lock);  
     //然后由配货系统进行处理.....等操作
    //3：把处理过的改为已处理状态
    $success = array(  
        ‘status‘=>1,  
        ‘updated_at‘=>date(‘Y-m-d H;i:s‘,time()),    
    );  
    $res_last = $db->update(‘order_queue‘,$success,$lock);  
    if($res_last){  
       echo "Success:".$res_last;   
    }else{  
        echo "Fail:".$res_last;  
    }  
}else{  
    echo "ALL Finished";  
}  
?>  

#!/bin/bash  
date "+%G-%m-%d %H:%M:%S"    //当前年月日
cd /data/wwwroot/default/mq/  
php goods.php  

*/1 * * * * /data/wwwroot/default/mq/good.sh >> /data/wwwroot/default/mq/log.log 2>&1 //指定脚本目录并格式化输出//当然要创建log.log文件

tail -f log.log  //监控日志

<?php
$db = new Db();
$sms = new Sms();
while(true){
    $item = $db->getFirstRecord(); //获取数据表第一条记录
    if(!$item){
        //如果队列中没有数据，则结束定时器
        break;
    }
    $res = $sms->send($item[‘phone‘],$item[‘content‘]); //发送短信
    if($res){
        $db->deleteFristRecord(); //删除发送成功的记录
        echo $item[‘phone‘].‘发送成功‘;
    }else{
        echo $item[‘phone‘].‘发送失败，稍后继续尝试‘；
    }
    sleep(10); //每隔十秒循环一次            
}
 
echo ‘发送完毕！‘;
?>

 CREATE TABLE `redis_queue` (
  `id` int(10) unsigned NOT NULL AUTO_INCREMENT,
  `uid` int(11) NOT NULL DEFAULT ‘0‘,
  `time_stamp` varchar(24) NOT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=88 DEFAULT CHARSET=utf8;

<?php

$redis = new Redis();
$redis->connect(‘127.0.0.1‘,6379);
$redis-_name = ‘miaosha‘;


//秒杀用户涌入模拟,100个用户
for ($i =0; $i < 100; $i++) {
    $uid = rand(1000000,99999999);
}
//检查redis链表长度（已存在数量)
$num = 10;
if ($redis->lLen($redis_name) < 10 ) {
    //加入链表尾部
    $redis->rPush($redis_name, $uid.‘%‘.microtime());
} else {  //如果达到10个
    //秒杀结束
}
$redis->close();

<?php
//从队列头部读一个值，判断这个值是否存在，如果存在则切割出时间、uid保存到数据库中。（对于redis而言，如果从redis取出这个值，那么这个值就不在redis队列里了，如果出现问题失败了，那么我们需要有一个机制把失败的数据重新放入redis链表中）
$redis = new Redis();
$redis->connect(‘127.0.0.1‘,6379);
$redis-_name = ‘miaosha‘;

//死循环检测redis队列
while(1) {
    $user = $redis->lpop($redis_name);
    if (!$user || $user == ‘null‘) {  //如果没有数据跳出循环
        //如果一直执行，速度是非常快的，那么服务器压力大，这里2秒一次
        sleep(2);
        //跳出循环
        continue;
    } 
    //拿出微秒时间戳和uid
    $user_arr = explode(‘%‘, $user);
    $insert_data = array(
        ‘uid‘ => $user_arr[0];
        ‘time_stamp‘ => $user_arr[1];
    );
    $res = $db->insert(‘redis_queue‘, $insert_data);
    //如果插入失败
    if (!$res) {
        //从哪个方向取出，从哪个方向插回
        $redis->lpush($redis_name, $user);
        sleep(2);
    }
} 
$redis->close();