博客
关于我
强烈建议你试试无所不能的chatGPT,快点击我
php队列使用
阅读量:5856 次
发布时间:2019-06-19

本文共 2413 字,大约阅读时间需要 8 分钟。

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

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

转载于:https://www.cnblogs.com/ngx171/p/8579359.html

你可能感兴趣的文章
Linux平台Java调用so库-JNI使用例子
查看>>
Spring Data JPA
查看>>
项目管理修炼之道之规划项目
查看>>
Web服务器压力测试工具http_load、webbench、ab、Siege使用教程
查看>>
RHEL6.3 源码安装Puppet
查看>>
Mac软件下载备忘
查看>>
java 泛型初探
查看>>
在Linux中执行.sh脚本,异常/bin/sh^M: bad interpreter: No such file or directory
查看>>
就是一个表格
查看>>
CakePHP 2.x CookBook 中文版 第三章 入门 之 CakePHP 的结构
查看>>
Objective-C的算术表达式 .
查看>>
找回使用Eclipse删除的文件
查看>>
rabbitmq 消息系统 消息队列
查看>>
集成spring3、hibernate4、junit
查看>>
URL与ASCII
查看>>
Redis.conf 说明
查看>>
我的友情链接
查看>>
java读取properties配置文件
查看>>
LVS+keepalived负载均衡
查看>>
UITableview中cell重用引起的内容重复的问题
查看>>