php队列使用

由于项目中在修改产品的同时要同步关联水单,删单，客保 等等数据。所以不可能等待所有都执行完毕以后再给客户端反馈。所以自己用写了个队列。在这里晒出来代码，以供大家参考。(项目中用到的是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(); ???}}

这里主要是实现异步处理，入队和出队解耦,单线程