~~~
<?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實現定時任務
- 檢測文件變化