由于项目中在修改产品的同时要同步关联水单,删单,客保 等等数据。所以不可能等待所有都执行完毕以后再给客户端反馈。所以自己用写了个队列。在这里晒出来代码,以供大家参考。(项目中用到的是tp,所以在这里用tp作为演示)
思路
1,需要用到队列则扔到queue表中。
2,利用linux计划任务 * * * * * 每分钟去执行 CronMission方法,在方法中控制执行频率 (分钟%5 则为每五分钟执行一次)
3利用 flock文件排它锁,保证单线程执行,避免重复执行
具体实现如下:
表:
CREATE TABLE IF NOT EXISTS `roav2_queue` (
`id` int(10) NOT NULL AUTO_INCREMENT,
`taskpath` varchar(50) NOT NULL DEFAULT '',//任务路径 `param` text NOT NULL,//参数在这里使用serialize编译 `status` tinyint(4) NOT NULL DEFAULT '0',//状态0未执行,1执行过 `created_at` datetime NOT NULL, `updated_at` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (`id`), KEY `created_at` (`created_at`)) ENGINE=InnoDB DEFAULT CHARSET=utf8 AUTO_INCREMENT=1 ;
1,获取队列
public function getQueueTask($limit = 10) { $map = ['status' => 0]; $data = $this->where($map)->order('id asc')->limit($limit)->select();// var_dump(M()->getLastSql());die; return $data; }
2,参数拼接
public function cmdHandle($phpCmd, $taskPath, $param){ $str = ''; if($param) { $params = unserialize($param); foreach ($params as $k => $v) { $str .= '/' . $k . '/' . $v; } $taskPath .= $str; } $res = $phpCmd . $taskPath; return $res;}
3,修改任务状态
public function changeTaskStatus($id, $status){ return save_r($this, ['id' => $id], ['status' => $status]);}
4,执行队列
public function doQueue(){ $phpCmd = C('PHPCMD'); $logPath = C('CRON_LOG_PATH'); $fp = fopen('flock.txt', 'w+'); if (flock($fp, LOCK_EX|LOCK_NB))//使用文件排他锁保证单线程执行 { $tasks = $this->getQueueTask(10); foreach ($tasks as $v) { $job = $this->cmdHandle($phpCmd, $v['taskpath'], $v['param']); $job .= ' >> ' . $logPath . 'doQueueMission' . date('Y-m-d') . '.log '; system($job); $this->changeTaskStatus($v['id'], 1); } flock($fp, LOCK_UN); } fclose($fp);}
5,添加任务
public function addTask($taskPath, $param){ if(empty($taskPath) || empty($param)) return 0; $data = [ 'taskpath' => $taskPath, 'param' => serialize($param), 'created_at' => date('Y-m-d H:i:s') ]; $res = add_r($this, $data); if ($res) return 1; else return 0;}
6,计划任务
public function cronMission(){ $minute = date('i'); //5分钟执行一次 if ($minute%5 == 0) { $this->model->doQueue(); }}
这里主要是实现异步处理,入队和出队解耦,单线程