>[danger] 20180415日進行了修改
>
>使用參見教程 高階篇七
>
~~~
<?php
/**
* Created by PhpStorm.
* User: Mikkle
* QQ:776329498
* Date: 2017/10/19
* Time: 10:10
*/
namespace mikkle\tp_worker;
use mikkle\tp_master\Db;
use mikkle\tp_master\Exception;
use mikkle\tp_master\Log;
use mikkle\tp_tools\Time;
abstract class CycleWorkBase
{
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
* @return \mikkle\tp_redis\Redis
*/
protected static function redis()
{
return WorkerRedis::instance();
}
/**
* @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=3600*24){
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::getDefaultTimeString(),
];
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 ();
}
}
~~~
- 序言及更新日志
- 前言一 開發PHP必備的環境(你可以不看)
- LinUX系統ThinkPHP5鏈接MsSQL數據庫的pdo_dblib擴展
- centos7.2掛載硬盤攻略
- Centos系統Redis安裝及Redis的PHP擴展安裝
- Centos系統增加Swap(系統交換區)的方法
- 前言二 開發PHP軟件配置和介紹(你依然可以不看)
- 數據庫SQL文件
- 本地Git(版本控制)的搭建
- GIT遠程倉庫的克隆和推送
- Git常用命令
- PHP面向對象思想實戰經驗領悟
- PHP面向對象實戰----命名空間
- PHP面向對象實戰----繼承
- 基類實戰--底層方法封裝
- 基類實戰--構造函數實戰
- 基類實戰--析構函數的使用
- TP5實戰開發前篇---控制器(controller)
- 控制器中Request類的使用
- 控制器中基類的使用
- TP5實戰開發前篇---模型篇(model)
- TP5實戰開發前篇---驗證器篇(Validate)
- TP5實戰課程入門篇---花拳繡腿
- 模塊以及類的文件的建立
- Api開發------單條信息顯示
- Api開發---單條信息復雜關聯顯示
- Api開發---查詢信息緩存Cache的應用
- TP5實戰技巧---開發思路 引路造橋
- TP5實戰技巧---整合基類 化繁為簡
- TP5實戰課程入門篇---數據操作
- Api開發---數據的添加和修改
- API開發---快速開發API通用接口
- TP5專用微信sdk使用教程
- THINKPHP5微信SDK更新記錄及升級指導
- TP5專用SDK 微信參數配置方法
- 微信公眾號推送接口對接教程
- 微信推送接口對接示例含掃描登錄微信端部分
- TP5專用微信支付SDK使用簡介
- TP5專用支付寶支付SDK使用說明
- 使用NW將開發的網站打包成桌面應用
- TP5高階實戰課程 進階篇概述
- 進階篇一 實戰開發之習慣及要求
- 進階篇二 實戰開發之控制器
- 控制器基類之控制器基類使用方法
- 控制器基類之控制器基類常用方法分享
- 控制器基類之構造函數的使用方法
- 進階篇三 實戰開發之權限控制
- TP5實戰源碼 --- 全局用戶信息驗證類Auth
- TP5實戰源碼 --- 微信Auth實戰開發源碼
- 進階篇四 實戰開發之模型
- 模型基類之模型基類的用途
- 模型基類之常用數據處理方法
- 模型邏輯層之實戰代碼(含事務)
- 模型實戰開發之模型常用方法
- 模型實戰源碼 --- 樂觀鎖的應用
- 模型實戰技巧---Model事件功能的使用
- 模型事件實戰應用---數據庫操作日志
- 進階篇五 實戰開發之緩存(Cache)
- TP5實戰源碼---應用緩存獲取城市信息
- TP5實戰源碼---應用緩存獲取分類詳情
- 進階篇六 TP5類庫的封裝和使用
- DataEdit快捷操作類庫
- ShowCode快捷使用類庫
- 阿里大于 短信API接口 TP5專用類庫
- DatabaseUpgrade數據庫對比及更新類庫
- AuthWeb權限類使用說明
- 進階篇七 服務層的應用
- 服務層源碼示例
- 服務層基類源碼
- 進階篇八 應用層Redis數據處理基類
- Redis服務層基類源碼
- 進階篇九 使用Redis類庫處理一般的搶購(秒殺)活動示例
- 進階篇十 某大型項目應用本Redis類源碼示例(含事務 樂觀鎖)
- 進階篇十一 邏輯層的應用
- 邏輯層基類源碼
- 進階篇 服務層代碼示例
- 高階實戰課程 進階篇持續新增中
- 高階篇一 TP5命令行之守護任務源碼
- TP5實戰源碼 --- 命令行
- TP5實戰源碼 --- 通過shell建立PHP守護程序
- 高階篇二 使用Redis隊列發送微信模版消息
- 高階篇二 之 Worker隊列基類源碼
- 高階篇三 TP5實戰之Redis緩存應用
- Redis實戰源碼之Hash專用類庫源碼
- Redis實戰源碼之Model類結合
- Redis實戰源碼之模型Hash基類源碼
- Redis實戰源碼之Hash查詢使用技巧
- Redis實戰源碼之 shell腳本中redis賦值和取值
- 高階篇四 Swoole的實戰應用
- swoole基類代碼
- Swoole擴展WebsocketServer專用類
- 基于Swoole的多Room聊天室的程序
- Swoole守護服務shell源碼
- 高階篇五 命令行異步多進程隊列類的應用
- tp_worker類源碼
- WorkerBase
- WorkerCommand
- WorkerRedis
- Redis類
- CycleWorkBase
- WorkerHookBase異步鉤子
- 隊列日志SQL
- 高階篇六 定時執行隊列類庫以及使用方法
- 定時隊列類庫源碼
- 高階篇七 異步執行循環隊列類庫以及使用教程
- CycleWorkBase源碼
- 高階實戰課程 進階篇持續新增中
- Extend便捷類庫源碼庫
- 阿里相關類庫
- SendSms--驗證碼API接口文件
- 權限相關類庫目錄
- AuthWeb 權限驗證類庫
- Redis便捷操作類庫(20171224更新)
- Redis
- Tools工具類庫集
- Curl類庫
- DataEdit
- Rand類庫
- ShowCode類庫
- Upload類庫
- 附件集合
- 附件一:微信支付 實戰開發源碼
- 微信支付類庫源代碼
- Common_util_pub.php
- DownloadBill_pub.php
- JsApi_pub.php
- NativeCall_pub.php
- NativeLink_pub.php
- OrderQuery_pub.php
- Refund_pub.php
- RefundQuery_pub.php
- SDKRuntimeException.php
- ShortUrl_pub.php
- UnifiedOrder_pub.php
- Wxpay_client_pub.php
- Wxpay_server_pub.php
- WxPayConf_pub.php
- 微信支付回調頁面源碼
- 附件二 順豐快遞BSP接口實戰開發源碼
- 順豐快遞BSP接口實戰開發源碼
- 順豐BSP基類
- 順豐BSP基礎代碼
- 順豐BSP下單接口
- 順豐BSP查單接口
- 順豐BSP確認/取消接口
- 附件三 APP注冊登陸接口源碼(含融云平臺接口)
- 附件四 TP5訂單Model(含事務 獲取器 修改器等方法)
- 附錄五 RSA加密解密
- Rsa文件源碼
- 附件六 阿里大于短信接口
- 附件七 AES加解密類
- AES加解密類源碼
- 附件八 TP5路由設置源碼
- 附件九 TP5 Excel導入導出下載便捷類庫
- Excel類庫TP5源碼
- 附件十 TP5便捷操作Redis類庫源碼
- TP5源碼 Redis操作便捷類庫
- 附件十一 TP5源碼 上傳文件入庫類源碼
- 上傳類Upload源碼
- Upload類上傳配置文件
- 存儲圖像文件的數據庫SQL文件
- 存儲文件的數據庫SQL文件
- 附件十二 TP5 圖片處理增強類 支持縮略圖在線顯示
- 附件十三 微信推送消息接口類庫源碼
- 附件十三 微信推送消息接口類庫源碼 之 基類
- 附件十四 存儲微信昵稱的處理方法