php获取后台Job管理的实现代码

从数据库中取出job 队列 创建新的进程进行执行 等待job 结束
代码如下:

1. <?php defined('SYSPATH') OR die('No direct access allowed.');
   
2. class Controller_Jobs extends Controller_Base{
   
3. public function before(){
   
4. parent::before();
   
5. if(Request::$protocol != "cli"){
   
6. die("Only cli allowed!\n");
   
7. }
   
8. }
   
9. public function after(){
   
10. parent::after();
    
11. //do some cleaning tasks
    
12. }
    
13. private function _execJobCommand($joburi,$paras){
    
14. $php_exec = Kohana::config("picsou.php_exec");
    
15. $php_index = APPINDEX;
    
16. $command_args = array();
    
17. $command_args[] = $php_index;
    
18. $command_args[] = "--uri=".$joburi;
    
19. foreach ($paras as $para => $value){
    
20. $command_args[] = "--".$para."=".$value;
    
21. }
    
22. //var_dump($command_args);exit;
    
23. echo "exec commmand:".$php_exec."\n";
    
24. pcntl_exec($php_exec,$command_args);
    
25. }
    
26. /*
    
27. * Running jobs in queues
    
28. */
    
29. public function action_run(){
    
30. $requestCount = 0;
    
31. while(true){
    
32. $sql = "select * from job_queue where status='1' and approved='1' order by id";
    
33. $jobs = DB::query(Database::SELECT,$sql)->execute()->as_array();
    
34. if($jobs){
    
35. foreach ($jobs as $job){
    
36. $requestCount ++;
    
37. //update the jobs status as running
    
38. DB::update('job_queue')->set(array('status'=>'2'))
    
39. ->where('id','=',$job['id'])->execute();
    
40. $job_pid = pcntl_fork();
    
41. if($job_pid == -1){
    
42. die("Could not fork Child");
    
43. } else if($job_pid == 0 ){
    
44. $this->_execJobCommand($job['job_uri'],json_decode($job['paras'],true));
    
45. echo "finish Child\n";
    
46. exit(0);
    
47. //run jobs here
    
48. } else{
    
49. echo "Waiting for job\n";
    
50. ob_flush();
    
51. $child_pid = pcntl_waitpid($job_pid,$status, WUNTRACED);
    
52. echo "waitpid end:".$status."\n";
    
53. if($status == 0){
    
54. //job completed
    
55. DB::update('job_queue')->set(array('status'=>'999'))
    
56. ->where('id','=',$job['id'])->execute();
    
57. echo "Child Finished\n";
    
58. ob_flush();
    
59. }else{
    
60. DB::update('job_queue')->set(array('status'=>'-1'))
    
61. ->where('id','=',$job['id'])->execute();
    
62. echo "Child Failed\n";
    
63. ob_flush();
    
64. }
    
65. }
    
66. }
    
67. }
    
68. else{
    
69. if($requestCount >=10){
    
70. echo "Have a rest, I have processed 10 jobs\n";
    
71. exit;
    
72. }
    
73. //no job to run
    
74. //echo "No job\n";
    
75. ob_flush();
    
76. sleep(5);
    
77. }
    
78. }
    
79. }
    
80. }

复制代码