原文详见:http://www.ucai.cn/blogdetail/7026?mid=1&f=12
可以在线运行查看效果哦!
五、队列具体实现二:定义调用接口
- <?php
- /**
- * 优才网公开课示例代码
- *
- * 任务队列实现
- *
- * @author 优才网全栈工程师教研组 张友林
- * @see http://www.ucai.cn
- */
- include_once('db.php');
- class Queue
- {
- /**
- * 把任务扔到队列
- *
- * @param string $taskphp 执行任务的程序
- * @param string $param 执行任务程序所用的参数
- * 例如,群发消息加入队列:
- * $arr = array(
- * "uid" => 4,//发信息的人的UID
- * "uids" => array(6,234,34,67,7888,2355), //接收信息的人的UID
- * "content" => 'xxxxx',//信息内容
- * );
- * $cqueue = new Queue();
- * $cqueue->add("/app/send_msg.php", serialize($arr));
- *
- */
- public function add($taskphp,$param)
- {
- $taskphp = mysql_real_escape_string($taskphp);
- //$param = mysql_real_escape_string($param);
- $param = $param;
- $sql = "insert into queue (taskphp, param) values('".$taskphp."', '".$param."')";
- $re = execute($sql);
- if ($re)
- {
- $pid = mysql_insert_id();
- return $pid;
- }
- else
- {
- return false;
- }
- }
- /**
- * 读取任务队列
- *
- * @param string $limit 一次取多少条
- */
- public function getQueueTask($limit = 1000)
- {
- $limit = (int)$limit;
- $sql = "select id, taskphp, param from queue where status = 0 order by id asc";
- $re = query($sql);
- return $re;
- }
- /**
- * 更新任务状态
- *
- * @param string $limit 一次取多少条
- */
- public function updateTaskByID($id)
- {
- $id = (int)$id;
- $mtime = time();
- $sql = "update queue set status =1, mtime = ".$mtime." where id = ".$id;
- $re = execute($sql);
- return $re;
- }
- public static function a2s($arr)
- {
- $str = "";
- foreach ($arr as $key => $value)
- {
- if (is_array($value))
- {
- foreach ($value as $value2)
- {
- $str .= urlencode($key) . "[]=" . urlencode($value2) . "&";
- }
- }
- else
- {
- $str .= urlencode($key) . "=" . urlencode($value) . "&";
- }
- }
- return $str;
- }
- public static function s2a($str)
- {
- $arr = array();
- parse_str($str, $arr);
- return $arr;
- }
- }
- ?>
- $cqueue = new Queue();
1:加入队列接口
//$param1 为执行任务的程序,$param2 为程序参数,可以为序列化的数据
$cqueue->add($param1,$param2);
2: 读取队列接口
$tasks = $cqueue->getQueueTask($limit = 1000);
3:更新任务状态
$cqueue->updateTaskStatus($id);
4:a2s是自定义的一个数组转换字符串方法,这里不要使用json_encode,容易出现问题,同样,从数据库中取出转换为数组的时候,使用s2a方法
$re = $cqueue->add("sendMsg.php", Queue::a2s($arr));
六、队列具体实现三:写执行队列的程序
根据设计,执行队列的程序文件是 do_queue.php , 它的主要功能是把任务从队列表里取出来,并且在后台执行。
do_queue.php部分代码:
- $phpcmd = exec("which php"); //查找到php安装位置
- $cqueue = new Queue();
- $tasks = $cqueue->getQueueTask(200);
- foreach ($tasks as $t)
- {
- $taskphp = $t['taskphp'];
- $param = $t['param'];
- $job = $phpcmd . " " . escapeshellarg($taskphp) . " " . escapeshellarg($param);
- system($job);
- }
七、具体任务的业务实现
还是拿群发消息来做例子,我们需要写好一个群发消息的程序,这个程序接收事先定义好的参数,然后根据参数调用发消息的接口把消息发送出去。
这个一般由做业务功能的工程师实现。但是架构师事先得写文档例子,教会别人使用。
- send_msg.php:
- $para = $argv[1];
- $arr = unserialize($para);
- $cmessage = new Message();
- foreach($arr['uids'] as $touid)
- {
- $cmessage->send($arr['uid'], $touid, $arr['content']);
- }
八 、服务器部署一:配置crontab
咱们执行队列的程序都写好了, 这个程序怎么触发呢,当然就要用到linux的定时任务,每隔一定的时间,执行do_queue.php一次。但是呢,这里不是直do_queue.php,咱们再提高一层,加个调度程序cron_mission.php, 在cron_mission.php里面调用do_queue.php。
配置定时任务 crontab:
crontab –e
* * * * * cd /ucai/schedule; php cron_mission.php >> cron_mission.log
#可以先使用crontab -l查看本机已经使用的定时任务
九、服务器部署二:写定时任务调度程序
思路:将定时任务写入到任务调度程序cron_mission.php中,这样可以在cron_mission.php中灵活控制队列任务。相比较直接通过crontab控制doQueue.php而言,避免了频繁修改服务器上的crontab,从安全,便于维护等角度来说,都是上策。
- cron_mission.php 示例:
- if ($minute % 5 == 0)
- {
- if(chdir($site_dir."app/")) {
- $cmd = "$phpcmd do_queue.php > do_queue.log &";
- echo '[' , $ymd , ' ' , $hour , ':' , $minute , '] ' , $cmd , "\n";
- system($cmd);
- }
- }
十、开启多进程并发执行队列
思路:对任务序列进行编号,数据库中执行的时候
where条件加上id%每个队列要执行任务总数=队列编号
这样可以避免重复处理
例如:每个进程执行10条任务,修改如下
1:定时任务的修改
修改前:
- if ($minute % 5 == 0)
- {
- if(chdir($site_dir."app/")) {
- $cmd = "$phpcmd do_queue.php > do_queue.log &";
- echo '[' , $ymd , ' ' , $hour , ':' , $minute , '] ' , $cmd , "\n";
- system($cmd);
- }
- }
修改后:
- if ($minute % 5 == 0)
- {
- for ($i=0; $i < 10; $i++) {
- $cmd = "$phpcmd doQueue.php 10 $i>> doQueueMission".date('Y-m-d').".log ";
- echo date("Y-m-d H:i:s") . "\t : " .$cmd."\n";
- system($cmd);
- }
- }
//每次进行10个进程,$i来区分是当前的进程标示
2:队列执行程序的修改
修改前:
- $phpcmd = 'D:\work\wamp\bin\php\php5.3.10\php ';
- $cqueue = new Queue();
- $tasks = $cqueue->getQueueTask(200);
修改后:
- $phpcmd = 'D:\work\wamp\bin\php\php5.3.10\php ';
- $total=$argv[1];
- $i=$argb[2];
- $cqueue = new Queue();
- $tasks = $cqueue->getQueueTask($total,$i,200);
3:取队列接口的修改
修改前:
- public function getQueueTask($limit = 1000)
- {
- $limit = (int)$limit;
- $sql = "select id, taskphp, param from queue where status = 0 order by id asc";
- $re = query($sql);
- return $re;
- }
修改后:
- public function getQueueTask($total,$i,$limit = 1000)
- {
- $limit = (int)$limit;
- $sql = "select id, taskphp, param from queue where status = 0 and id%$total=$i order by id asc";
- $re = query($sql);
- return $re;
- }
4:需要关注服务器压力
进程数定为多少,取决于服务器压力
十一、实现任务优先级
1:任务存储表加优先级字段
在数据表里,加一个优先级字段,按字段值的数值大小来区分优先级。
2:修改取队列任务接口,按优先级取
同样是在sql语句中增加order by。
十二、记录队列日志
1:关键地方加echo
2:shell脚本的>>和>的各自作用
总结:
我们这里的队列实现借助了服务器的计划任务来实现,例如linux中的crontab,这本身是linux系统中的一个程序,平时我们还可以使用他来进行定时执行.sh脚本,例如将数据库备份打包并ftp传送到指定服务器上,这个功能不需要借助php脚本,直接用.sh脚本就可以实现。在这里我们巧妙的将crontab和php脚本结合,并且使用crontab来不断调用一个队列调度接口cronMission.php,再通过cronMission.php直接来控制具体什么时候或者是满足什么条件来执行什么队列任务。
这里面几个需要注意的地方
1:往数据库中存取数据时,不要直接使用json_encode或者json_decode,容易造成一些意外问题,在代码中,我们定义了a2s和s2a两个方法,分别是处理数组转为字符串,和从数据库中读取字符串后转为数组。
2:当任务量比较大,同时服务器负载又没有充分利用的时候,可以使用多进程并发处理,在并发处理的时候需要考虑一个问题,就是如何避免重复,在这里我们使用了,对队列任务进行标记,每次从数据库中读取一个进程需要处理的一批任务,使用数据库中id与批次标示取余等于0的方法来区分,避免不同批次的队列,重复处理相同任务。(上面步骤10中有具体实现)
相关推荐
1、什么是死信队列 2、死信队列应用场景 3、延迟消息概念 4、死信队列使用流程图 5、死信队列应用(管理控制台) 6、死信队列应用(Java代码)
野火开发板,基于FreeRTOS的消息队列源代码。
主要介绍了剖析Java中阻塞队列的实现原理及应用场景,这里也对阻塞和非阻塞队列的不同之处进行了对比,需要的朋友可以参考下
【课程简介】 消息队列是个高大上的名词儿,本质上就是如何业务进行...了解这类消息系统使用的步骤和应用场景。 4-1 RabbitMQ (13:41) 第5章 总结 综述消息队列在各种场景使用的方法和注意事项。 5-1 总结.mp4 (02:48)
栈与队列的应用与区别
描述ucOS消息队列的使用,此文网络截文,感谢原创作者。
在目前广泛的Web应用中,都会出现一种场景:在某一个时刻,网站会迎来一个用户请求的高峰期(比如:淘宝的双十一购物狂欢节,12306的春运抢票节等),一般的设计中,用户的请求都会被直接写入数据库或文件中,在高...
关于消息队列的使用 一、消息队列概述 消息队列中间件是分布式系统中重要的组件,主要解决...二、消息队列应用场景 以下介绍消息队列在实际应用中常用的使用场景。异步处理,应用解耦,流量削锋和消息通讯四个场景。
第三节课:Rabbitmq镜像队列搭建 开发应用场景一:集群节点安装1.1)集群节点安装(老师的是192.168.159.86;在这里老师是以我的86服务器单台
介绍了消息队列的定义、为什么需要消息队列,应用场景,消息队列的优缺点
1、master选举:mastersel 2、数据的发布和订阅:subscribe 3、负载均衡:balance 4、分布式锁:lock 5、分布式队列:queue 6、命名服务:nameservice 演示代码下载(代码来自极客学院演示demo):
延迟队列的应用场景: 1、新用户注册,10分钟后发送邮件或站内信。 2、用户下单后,30分钟未支付,订单自动作废。 我们通过redis的有序集合zset来实现简单的延迟队列,将消息数据序列化,作为zset的value,把消息...
9、NIFI综合应用场景-通过NIFI配置kafka的数据同步 网址:https://blog.csdn.net/chenwewi520feng/article/details/130622776 本文旨在介绍nifi与kafka的交互过程,即生产数据到kafka中,然后通过nifi消费kafka中的...
优先队列适用于多种实际应用场景,如优先级调度、网络流控、事件处理、多线程同步和数据挖掘等。在实际应用中,优先队列可以大大提高我们的工作效率,为我们的工作和生活带来便利。 总的来说,优先队列是一种强大的...
主要介绍了基于ZooKeeper实现队列源码的相关内容,包括其实现原理和应用场景,以及对队列的简单介绍,具有一定参考价值,需要的朋友可以了解下。
1、master选举:mastersel 2、数据的发布和订阅:subscribe 3、负载均衡:balance 4、分布式锁:lock 5、分布式队列:queue 6、命名服务:nameservice 资料来自极客学院
消息队列有哪些使用场景? 为什么使用消息队列进行应用解耦? 为什么使用消息队列进行异步处理? 为什么使用消息队列进行流量消峰? 为什么使用消息队列进行消息通信? 如何使用消息队列进行日志处理? ...............
基于WCF分布式消息队列,可以在一些客户端并不需要服务端响应的场景上应用。消息队列在异步处理上有巨大优势,是一项可选择性的进程间的通信,能够保持进程之间通信的稳定性。
Work Queues与入门程序的简单模式相比,多了一个或一些消费端,多个消费端共同消费同一个队列 中的消息。 应用场景:对于任务过重或任务较多情况使用工作队列可以提高任务处理的速度。