csapi/application/monitor/controller/Requestpop.php

<?php
namespace app\monitor\controller;
use app\common\service\MailService;
use think\Cache;
use think\Controller;
use think\Db;
use think\Log;
use think\Validate;

/**
 * 日志和邮件出队
 * Class RequestPop
 */
class Requestpop extends Controller
{

    private $redisClient = null;
    private $listMaxAlertNum = 500;
    private $emailMaxAlertNum = 50;

    /**
     * 请求队列
     * @throws \think\exception\PDOException
     */
    public function requestList(){
        
        try{
                $date = date('Y-m-d H:i:s');
                //切换当前用户，以免root权限日志
                $this->setUser();
                $pid = posix_getpid();
                //连接redis
                $this->connectionRedis();
                $remainNum = $this->redisClient->lLen('request_list');
                if($remainNum > $this->listMaxAlertNum){
                    $mailService = new MailService();
                    $res = $mailService->send('谢鹏','请求日志自动升级进程数',"剩余数量$remainNum,主动清理程序",[],[['mail'=>'2990412861@qq.com','name'=>'leo.xie']]);
                    Log::record("请求日志自动升级进程数".$res."\n");
                    //自动升级进程数
                    $requestPidList = Cache::get('request_list_pop_program');
                    is_array($requestPidList)?$requestPidList[] = $pid:$requestPidList = [$pid];
                    Cache::set('request_list_pop_program',$requestPidList);
                }else{
                    $requestPidList = Cache::get('request_list_pop_program');
                    if(is_array($requestPidList) && count($requestPidList) > 1){

                        Cache::rm('request_list_pop_program');
                        $mailService = new MailService();
                        $res = $mailService->send('谢鹏','请求日志自动降级进程数',"剩余数量$remainNum,主动清理程序",[],[['mail'=>'2990412861@qq.com','name'=>'leo.xie']]);
                        Log::record("请求日志自动降级进程数".$res);
                    }elseif(!empty($requestPidList) ){
                        die('['.$date.']有正在执行中的程序');
                    }
                    Cache::set('request_list_pop_program',[$pid]);
                }
                
                while(true){
                    
                    //有时候进程没有杀掉，做一次自我修复机制
                    $curPid = posix_getpid();
                    $requestPidList = Cache::get('request_list_pop_program');
                    if(!is_array($requestPidList) || !in_array($curPid, $requestPidList)){
                        Log::record('异常中断'.$curPid);
                        exit(0);
                    }
                    
                    $requestJson = $this->redisClient->lPop('request_list');
                    
                    if(empty($requestJson)){
                        //Log::record('['.$date.']我在等待执行...');
                        sleep(10);
                        //关闭之后，重新连接(防止异常险情)
                        try{
                            $this->redisClient->close();
                        }catch (\Exception $ex){
                            //nothing to do...
                        }
                        $this->connectionRedis();
                        continue;
                    }

                    $requestArr = @json_decode($requestJson,true);
                    if(empty($requestArr)){
                        continue;
                    }

                    $rule = [
                        'companyCode'=>'require',
                        'projectCode'=>'require',
                        'requestUrl'=>'require',
                        'requestParam'=>'require',
                        'returnCode'=>'require',
                        'returnMsg'=>'require',
                        'returnData'=>'require',
                        'responseTime'=>'require',
                        'requestTime'=>'require',
                    ];

                    $validate = new Validate($rule);
                    if(!$validate->check($requestArr)){
                        continue;
                    }

                    //进入处理程序
                    $dbData = [
                        'company_code'=>$requestArr['companyCode'],
                        'project_code'=>$requestArr['companyCode']."_".$requestArr['projectCode'],
                        'request_url'=>$requestArr['requestUrl'],
                        'request_param'=>  json_encode($requestArr['requestParam']),
                        'return_code'=>$requestArr['returnCode'],
                        'return_msg'=>$requestArr['returnMsg'],
                        'return_data'=>  json_encode($requestArr['returnData']),
                        'response_time'=>$requestArr['responseTime'],
                        'request_time'=>$requestArr['requestTime'],
                        'insert_time'=>date('Y-m-d H:i:s'),
                        'date_range'=>date('Y-m-d H:i',intval(time()/300)*300)
                    ];

                    $Ymd = date('Ymd');
                    try {
                        Db::connect('monitor')->table("api_request_$Ymd")->insert($dbData);
                    }catch (\Exception $ex){

                        if(strpos($ex->getMessage(),'Base table or view not found') !== false){
                            Db::connect('monitor')->execute("CREATE TABLE api_request_$Ymd LIKE api_request");
                            Db::connect('monitor')->table("api_request_$Ymd")->insert($dbData);
                        }
                    }
                }
        }catch(\Exception $ex){
            //自动降级进程数
            echo $ex->getMessage()."\n";
            $this->restart(2);
            echo "cache\n";
            Cache::rm('request_list_pop_program');
            exit(0);
        }
    }


    public function emailList(){
        
        try{
                $date = date('Y-m-d H:i:s');
                //切换当前用户，以免root权限日志
                $this->setUser();
                $pid = posix_getpid();
                //连接redis
                $this->connectionRedis();
                $remainNum = $this->redisClient->lLen('email_list');
                if($remainNum > $this->emailMaxAlertNum){
                    $mailService = new MailService();
                    $res = $mailService->send('谢鹏','报警日志自动升级进程数',"剩余数量$remainNum,主动清理程序",[],[['mail'=>'2990412861@qq.com','name'=>'leo.xie']]);
                    Log::record("报警日志自动升级进程数".$res);
                    //自动升级进程数
                    $emailPidList = Cache::get('email_list_pop_program');
                    is_array($emailPidList)?$emailPidList[] = $pid:$emailPidList = [$pid];
                    Cache::set('email_list_pop_program',$emailPidList);
                    
                }else{
                    $emailPidList = Cache::get('email_list_pop_program');
                    if(is_array($emailPidList) && count($emailPidList) > 1){
                        Cache::rm('email_list_pop_program');
                        $mailService = new MailService();
                        $res = $mailService->send('谢鹏','报警日志自动降级进程数',"剩余数量$remainNum,主动清理程序",[],[['mail'=>'2990412861@qq.com','name'=>'leo.xie']]);
                        Log::record("报警日志自动降级进程数".$res);
                    }elseif(!empty($emailPidList)){
                        die('['.$date.']有正在执行中的程序');
                    }
                    
                    Cache::set('email_list_pop_program',[$pid]);
                }

                while(true){
                    $emailJson = $this->redisClient->lPop('email_list');
                    
                    //有时候进程没有杀掉，做一次自我修复机制
                    $curPid = posix_getpid();
                    $requestPidList = Cache::get('email_list_pop_program');
                    if(!is_array($requestPidList) || !in_array($curPid, $requestPidList)){
                        Log::record('异常中断'.$curPid);
                        exit(0);
                    }
                    
                    if(empty($emailJson)){
                        try{
                            //记录，每5分钟发送一次邮件
                            $sendEmailTs = Cache::get('send_email_ts');
                            $currentTs = time();
                            if($currentTs - intval($sendEmailTs)>300){
                                $companyList = Db::connect('monitor')->table('company')->where(['status'=>1])->cache(3600)->select();
                                if(!empty($companyList)){
                                    foreach($companyList as $company){
                                        $this->sendMail($company['company_code'], @explode(',', $company['email']));
                                    }
                                }
                                
                                Cache::set('send_email_ts',$currentTs);
                            }
                            $this->redisClient->close();
                        }catch (\Exception $ex){
                            //nothing to doing
                        }
                        sleep(20);
                        $this->connectionRedis();
                        continue;
                    }

                    $emailArr = @json_decode($emailJson,true);
                    if(empty($emailArr)){
                        continue;
                    }

                    $rule = [
                        'companyCode'=>'require',
                        'projectCode'=>'require',
                        'requestUrl'=>'require',
                        'requestParam'=>'require',
                        'msg'=>'require',
                        'line'=>'require',
                        'path'=>'require',
                        'requestTime'=>'require',
                    ];

                    $validate = new Validate($rule);
                    if(!$validate->check($emailArr)){
                        continue;
                    }

                    //进入处理程序
                    $dbData = [
                        'company_code'=>$emailArr['companyCode'],
                        'project_code'=>$emailArr['companyCode']."_".$emailArr['projectCode'],
                        'request_url'=>$emailArr['requestUrl'],
                        'request_param'=>  json_encode($emailArr['requestParam']),
                        'msg'=>$emailArr['msg'],
                        'line'=>$emailArr['line'],
                        'path'=>$emailArr['path'],
                        'request_time'=>$emailArr['requestTime'],
                        'insert_time'=>date('Y-m-d H:i:s'),
                        'json_data'=>$emailJson,
                        'is_send'=>0
                    ];

                    $Ymd = date('Ymd');
                    try {
                        Db::connect('monitor')->table("abnormal_request_$Ymd")->insert($dbData);
                    }catch (\Exception $ex){

                        if(strpos($ex->getMessage(),'Base table or view not found') !== false){
                            Db::connect('monitor')->execute("CREATE TABLE abnormal_request_$Ymd LIKE abnormal_request");
                            Db::connect('monitor')->table("abnormal_request_$Ymd")->insert($dbData);
                        }
                    }
                }
        }catch(\Exception $ex){
                //echo $ex->getMessage()."\n";
                $this->restart(1);
                //echo "cache\n";
                Cache::rm('email_list_pop_program');
                exit(0);
        }
    }

    private function connectionRedis(){
        $this->redisClient = new \Redis();
        $this->redisClient->connect('47.97.187.118', 6379);
        $this->redisClient->auth('gudong-hz');
    }
    
    /**
     * 发送对应的公司的客户列表
     * @param type $companyCode
     * @return boolean
     */
    private function sendMail($companyCode,$mailList){
        
        $Ymd = date('Ymd');
        
        $sendEmailList = Db::connect('monitor')->table("abnormal_request_$Ymd")->where(['is_send'=>0,'company_code'=>$companyCode])->select();
        if(empty($sendEmailList)){
            return true;
        }
        $sendContent = "";
        $oli = 1;
        $updateWhere = [];
        foreach($sendEmailList as $itemEmail){
            $updateWhere[] = $itemEmail['id'];
            $sendContent .= "<pre>";
            foreach($itemEmail as $key=>$eitm){
               if(in_array($key, ['request_param','json_data'])){
                   $itemEmail[$key] = @json_decode($eitm);
               }
            }
            $sendContent.= @json_encode($itemEmail,JSON_PRETTY_PRINT);
            $sendContent.="</pre><br/>===============".($oli++)."===================<br/>";
        }
        $mailService = new MailService();
        $sendMailList = [];
        foreach($mailList as $mailItem){
            $sendMailList[] = ['mail'=>$mailItem,'name'=>$mailItem];
        }
        
        if($sendMailList){
            $res = $mailService->send('谢鹏',"程序预警$companyCode",$sendContent,[],$sendMailList);
            Log::record("邮件".$res."\n");
        }
        
        try{
            Db::connect('monitor')->table("abnormal_request_$Ymd")->where(['is_send'=>0,'id'=>['in',$updateWhere]])->update(['is_send'=>1,'sender'=>"811329263@qq.com",'receiver'=> @implode(',', $mailList),'send_time'=>date('Y-m-d H:i:s'),'send_msg'=>'true']);
        }catch(\Exception $ex){
            Log::record("邮件异常".$ex->getMessage()."\n");
        }
    }

    //设置运行的用户
    private function setUser($name='www') {

        $maxPid = posix_getpid();
        $pid = pcntl_fork();
         if ($pid != 0) {
              posix_kill($maxPid, SIGKILL);
         }
         
        if (empty($name)) {
            return false;
        }
        
        $user = posix_getpwnam($name);
        if ($user) {
            $uid = $user['uid'];
            $gid = $user['gid'];
            $result = posix_setuid($uid);
            posix_setgid($gid);
        }
        return $result;
    }
    
    private function restart($flag=3){
        
        if($flag&1){
            Cache::rm('email_list_pop_program');
        }

        if($flag&2){
            Cache::rm('request_list_pop_program');
        }
    }
    
    
    public function restartAll(){
         Cache::rm('send_email_ts');
        $this->restart(3);
    }
    
    public function getCache(){
        echo "email_list_pop_program\n";
        $emailPidList = Cache::get('email_list_pop_program');
        var_dump($emailPidList);
        
        echo "request_list_pop_program\n";
        $requestPidList = Cache::get('request_list_pop_program');
        var_dump($requestPidList);
        
        echo "sendTs\n";
        $sendTs = Cache::get('send_email_ts');
        var_dump($sendTs);
    }
}