~~~
<?php
/**
* @Author: 陈静
* @Date: 2018/04/26 22:17:20
* @Description:
*/
namespace app\worker\controller;
use app\base\controller\Redis;
use think\Config;
use think\Db;
use think\Exception;
use think\Log;
abstract class CycleWorkerBase
{
protected $listName;
protected $redis;
protected $workList;
protected $workerName;
public static $instance;
protected $connect =[] ;
protected $saveLog=false;
protected $tableName = "mk_log_service_queue";
protected $error;
protected $stopTime;
protected $startTime;
protected $nextTime=60;
/**
* Base constructor.
* @param array $options
*/
public function __construct($options=[])
{
$this->_initialize();
$this->redis = $this->redis();
$this->workList = "worker_list";
$this->workerName = get_called_class();
$this->listName = md5($this->workerName);
}
public function _initialize(){
}
abstract protected function runCycleHandle($data);
protected function runHandle($data)
{
try{
$instance=self::instance();
$i =0;
while ( true ){
echo self::checkWorkingStop() ;
if ( self::checkWorkingStop() ){
echo "任务被终止!";
break;
}
$i++;
self::signWorking();
$time = $instance->getNextRunTime();
if (!empty($instance->startTime ) && is_int( $instance->startTime ) ){
if ( time()>strtotime( $instance->startTime )){
$instance->pcntlWorker($data,"runCycleHandle");
Log::notice("执行了程序pcntlWorker".$this->workerName );
}else{
$this->sleep($time);
Log::notice("未到执行时间_".$instance->startTime);
continue;
}
}else{
$instance->pcntlWorker($data,"runCycleHandle");
$this->sleep($time);
Log::notice("执行了程序pcntlWorker".$this->workerName );
}
Log::notice("睡眠$time 秒,循环执行程序执行程序".$this->workerName );
echo "睡眠$time 秒,循环执行程序执行程序.$this->workerName ";
}
$this-> clearWorkingWork();
Log::notice("循环执行程序执结束");
}catch (Exception $e){
Log::notice( $e ->getMessage());
}
}
protected function getNextRunTime(){
if (!empty( $this->nextTime ) && is_int( $this->nextTime)){
return $this->nextTime;
}
return 10;
}
/**
* @title redis
* @description redis加载自定义Redis类
* User: Mikkle
* QQ:776329498
*/
protected static function redis()
{
$options = empty($options) ? $redis = Config::get("redis") : $options;
return Redis::instance($options);
}
/**
* @title runWorker
* @description 标注命令行执行此任务
* User: Mikkle
* QQ:776329498
* @param string $handleName
*/
public function runWorker($handleName="run"){
$this->redis->hset($this->workList,$this->workerName,$handleName);
}
/**
* 标注命令行清除此任务
* Power: Mikkle
* Email：776329498@qq.com
*/
public function clearWorker(){
$this->redis->hDel("cycle_list",$this->listName);
$this->redis->hdel($this->workList,$this->workerName);
}
/**
* Power: Mikkle
* Email：776329498@qq.com
* @return static
*/
static public function instance(){
if (self::$instance){
return self::$instance;
}
return new static();
}
/**
*
* 当命令行未运行 直接执行
* description add
* User: Mikkle
* QQ:776329498
* @param $data
* @return bool
*/
static public function start($data=[]){
try{
$data=json_encode($data);
$instance = static::instance();
self::clearWorkingStop();
switch (true){
case (self::checkWorking()):
Log::notice("Work service is Running!!");
return "CycleWork service is Running!!";
break;
case (self::checkCommandRun()):
$instance->redis->hSet("cycle_list",$instance->listName,$data);
Log::notice("Command service start work!!");
$instance->runWorker();
break;
default:
Log::notice("Command service No away!!");
return "Command service No away!!";
// $instance->runHandle($data);
}
return "true";
}catch (Exception $e){
Log::error($e->getMessage());
return false;
}
}
/**
*
* 当命令行未运行 直接执行
* description add
* User: Mikkle
* QQ:776329498
*/
static public function stop(){
return self::signWorkingStop();
}
static public function status(){
return self::checkWorking();
}
static public function signWorking($time=20){
self::redis()->set(self::instance()->workerName."_run","true",$time);
}
static public function checkWorking(){
return self::redis()->get(self::instance()->workerName."_run") ? true :false;
}
static public function clearWorkingWork(){
self::redis()->hDel("cycle_list",self::instance()->listName);
return self::redis()->delete( self::instance()->workerName."_stop");
}
static public function signWorkingStop($time= 86400){
self::redis()->hDel("cycle_list",self::instance()->listName);
return self::redis()->set(self::instance()->workerName."_stop","true",$time);
}
static public function clearWorkingStop(){
return self::redis()->delete( self::instance()->workerName."_stop");
}
static public function checkWorkingStop(){
if (!empty(self::instance()->stopTime )){
if ( time()>strtotime( self::instance()->stopTime )){
return true;
}else{
return self::redis()->get(self::instance()->workerName."_stop") ? true :false;
}
}
return self::redis()->get(self::instance()->workerName."_stop") ? true :false;
}
/**
* 命令行执行的方法
* Power: Mikkle
* Email：776329498@qq.com
*/
static public function run(){
$instance = static::instance();
try {
$redisData = $instance->redis->hGet("cycle_list", $instance->listName);
if ($redisData) {
$data = json_decode($redisData, true);
if ($data) {
echo "开始执行循环任务" . PHP_EOL;
$instance->pcntlWorker($data);
$instance->redis->hDel("cycle_list", $instance->listName);
} else {
$instance->clearWorker();
}
} else {
$instance->clearWorker();
echo "未检测到循环任务" . PHP_EOL;
}
} catch (Exception $e) {
Log::error($e->getMessage());
echo($e->getMessage());
}
}
/**
* 检测命令行是否执行中
* Power: Mikkle
* Email：776329498@qq.com
* @return bool
*/
static public function checkCommandRun(){
return self::redis()->get("command") ? true :false;
}
public function getError(){
if (is_array($this->error )){
return json_encode( $this->error );
}
return $this->error;
}
/*
* 检查是注重某些值是非为空
*/
protected function checkArrayValueEmpty($array,$value,$error=true){
switch (true){
case (empty($array)||!is_array($array)):
if ($error==true){
$this->addError("要检测的数据不存在或者非数组");
}
return false;
break;
case (is_array($value)):
foreach ($value as $item){
if (!isset($array[$item]) || (empty($array[$item]) && $array[$item]!==0)){
if ($error==true) {
$this->addError("要检测的数组数据有不存在键值{$item}");
}
return false;
}
}
break;
case (is_string($value)):
if (!isset($array[$value]) || empty($array[$value] && $array[$value]!==0)){
if ($error==true) {
$this->addError("要检测的数组数据有不存在键值{$value}");
}
return false;
}
break;
default:
}
return true;
}
public function addError($error){
$this->error = is_string($error) ? $error : json_encode($error);
}
protected function saveRunLog($result,$data){
try{
$operateData = [
"class" => $this->workerName,
"args" => json_encode($data),
"result"=> $result ? "true":"false",
"error" => $this->error ? $this->getError() : null,
"time" => time(),
];
Db::connect($this->connect)->table($this->tableName)->insert($operateData);
}catch (Exception $e){
Log::error($e->getMessage());
}
}
protected function sleep($time=1){
if ($time<=10){
sleep(sleep($time));
}else{
$ci = $time/10;
for($i=0; $i<$ci;$i++){
self::signWorking();
echo "已睡眠 10 秒";
sleep(10);
}
sleep(sleep($time%10));
}
}
/**
* title 分进程
* description pcntlWorker
* User: Mikkle
* QQ:776329498
* @param $data
* @param string $action
*/
protected function pcntlWorker($data,$action = "runHandle")
{
try{
// 通过pcntl得到一个子进程的PID
$pid = pcntl_fork();
if ($pid == -1) {
// 错误处理：创建子进程失败时返回-1.
die ('could not fork');
} else if ($pid) {
// 父进程逻辑
// 等待子进程中断，防止子进程成为僵尸进程。
// WNOHANG为非阻塞进程，具体请查阅pcntl_wait PHP官方文档
pcntl_wait($status, WNOHANG);
} else {
// 子进程逻辑
$pid_2 = pcntl_fork();
if ($pid_2 == -1) {
// 错误处理：创建子进程失败时返回-1.
die ('could not fork');
} else if ($pid_2) {
// 父进程逻辑
echo "父进程逻辑开始" . PHP_EOL;
// 等待子进程中断，防止子进程成为僵尸进程。
// WNOHANG为非阻塞进程，具体请查阅pcntl_wait PHP官方文档
pcntl_wait($status, WNOHANG);
echo "父进程逻辑结束" . PHP_EOL;
} else {
// 子进程逻辑
echo "子进程逻辑开始" . PHP_EOL;
$this->$action( $data );
echo "子进程逻辑结束" . PHP_EOL;
$this->pcntlKill();
}
$this->pcntlKill();
}
}catch (Exception $e){
Log::error($e->getMessage());
}
}
/**
* Kill子进程
* Power: Mikkle
* Email：776329498@qq.com
*/
protected function pcntlKill(){
// 为避免僵尸进程，当子进程结束后，手动杀死进程
if (function_exists("posix_kill")) {
posix_kill(getmypid(), SIGTERM);
}
system('kill -9 ' . getmypid());
exit ();
}
}
~~~
- PHP7新特性
- 优雅的写代码
- 常见的代码优化
- 常用的工具类
- PHP原生生成EXCEL
- PHP地理位置计算
- PHP获取服务器状态
- 驼峰转下划线
- 百度地图两点坐标距离计算
- 判断是否是url
- PHP常见header头
- 邮件发送类
- 阿拉伯数字转化为大写
- 获取汉字首个拼音
- 根据身份证号获取星座
- 生成验证码类
- 生成唯一ID
- 身份证验证类
- PHP中文转拼音
- Nginx配置文件
- curl获取网页内容
- 快递查询api
- 上传图片类
- 股票类
- 找回密码类
- 字符串助手函数
- 校验数据规则
- PHP获取收集相关信息
- 字符串截取助手函数
- 网页中提取关键字
- 检测浏览器语言
- 微信相关类
- 微信获取access_token
- 获取用户基本信息
- 代码规范
- 编程规范(psr-1,2)
- 编程规范(原作者的建议)
- 经验
- 常用函数地址
- 函数集合
- 一些常识
- MYSQL相关知识
- 常用sql
- mysql事务隔离级别
- Read uncommitted
- Read committed
- Repeatable read
- Serializable
- 高性能MYSQL读书笔记
- 第一章MYSQL的架构
- mysql逻辑架构
- redis相关知识
- 1.安装redis
- 3.php操作redis
- 队列
- 悲观锁
- 乐观锁
- 发布
- 订阅
- redis实战-文章投票
- 设计模式
- 创建模型实例
- 单例模式
- 工厂模式
- AnimalInterface.php
- Chicken.php
- Factory.php
- Farm.php
- Pig
- SampleFactory.php
- Zoo
- 抽象工厂模式
- AnimalFactory
- Factory
- FarmInterface
- Income
- PandaZoo
- PeonyZoo
- PigFarm
- PlantFactory
- RiceFarm
- ZooInterface
- 原型模式
- 建造者模式
- 结构型模式实例
- 桥接模式
- 享元模式
- 外观模式
- 适配器模式
- 装饰器模式
- 组合模式
- 代理模式哦
- 过滤器模式
- 行为型模式实例
- 模板模式
- 策略模式
- 状态模式
- 观察者模式
- 责任链模式
- 访问者模式
- 解释器模式
- 空对象模式
- 中介者模式
- 迭代器模式
- 命令模式
- 备忘录模式
- 网络知识
- 互联网协议概述
- nginx简易交互过程
- HTTP知识
- LINUX相关知识
- swoole学习
- 1.初识swoole
- 2.WebSocket PHP 即时通讯开发
- 3.异步多进程的 CURL
- 4.异步非阻塞多进程的 Http 服务器
- 5.TCP 服务器
- 5.1同步 TCP 客户端
- 5.2异步 TCP 客户端
- 6.UDP 服务器
- 7.异步多任务处理
- 8.毫秒定时器
- 9.高并发投票
- ThinkPHP5学习
- 命令行操作
- 所有命令行中用到的基类
- 1.base
- 2.WorkerBase
- 3.TimeWorkerBase
- 4.CycleWorkerBase
- 5.WorkerCommandBase
- 6.WorkerHookBase
- 1.基础命令实现
- 2.建立Linux上的守护源码
- 3.发送模板消息
- 4.基于命令行实现自己的队列模式
- 5.发送定时短信
- thinkphp5使用sentry
- sentry通知，记录日志
- 高级查询
- Kafka相关
- 1.安装
- 2.为php打扩展
- 3.kafka实现
- 一些工具搭建
- sentry日志收集系统搭建
- walle搭建
- php实现定时任务
- 检测文件变化