由於項目中在修改產品的同時要同步關聯水單,刪單，客保 等等數據。所以不可能等待所有都執行完畢以後再給客戶端反饋。所以自己用寫了個隊列。在這裏曬出來代碼，以供大家參考。(項目中用到的是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();
    }
}

這裏主要是實現異步處理，入隊和出隊解耦,單線程

php隊列使用