knowledge/queue/base/dealBase.php

<?php
/**
 * 队列进程基类--生成子进程
 *
 * @version 2.0 - 2011-11-25
 * @package Daemon
 */

set_time_limit(0);

include_once(dirname(dirname(dirname(__FILE__)))."/library/publicBase.php");
abstract class dealBase extends publicBase {
    protected $processnum = 1;      // 程序限制进程数，默认为1
    protected $is_while = true;     // 是否无限循环，默认为是
    protected $deal_sleep = 1;      // 每执行一次后等待的时间，单位为妙，默认为1

    private $path_conf;             // 配置文件路径
    private $path_proc;             // 进程状态路径
    private $conf_md5;              // 配置文件MD5
    private $deal_md5;              // 执行队列文件MD5
    private $base_md5;              // 基础类文件MD5
    private $file_deal;             // 队列文件路径
    private $path_deal;             // 队列文件路径(包含标识)
    private $deal_flag='';          // 队列类别标识
    private $proc_code;             // 当前队列自定义进程数号码标识
    private $close_path;            // 开关标识文件

    public function __construct() {
        $this->path_conf = dirname(dirname(__FILE__))."/config/daemonconf.php";
        $this->path_proc = dirname(dirname(__FILE__)).'/cache/proc/deal_%s.txt';
        $this->close_path = dirname(dirname(__FILE__)).'/cache/proc/close.txt';

        $this->file_deal = $_SERVER['SCRIPT_FILENAME'];

        // 识别队列自定义进程数号码标识
        if(count($_SERVER['argv'])==3) {
            $this->deal_flag = $_SERVER['argv'][1];
            $this->proc_code = $_SERVER['argv'][2];
        } else {
            $this->proc_code = $_SERVER['argv'][1];
        }

        $this->path_deal = trim($this->file_deal.' '.$this->deal_flag);

        include_once($this->path_conf);

        $this->setPara();
        if($this->processnum+0==0) $this->processnum = 1;

        $this->collectVerifyMd5();
        do {
            $this->setProcHealth();
            if(file_exists($this->close_path)) {
                sleep(10);
                return ;
            }

            $this->checkVerifyMd5();
            //if(!$this->cProcessNum()) exit("Process limit <{$this->processnum}>\n");  // 每次循环执行任务调用ps系统命令导致cpu使用过高
            $this->deal();

            sleep($this->deal_sleep);
        } while($this->is_while);
        return;
    }

    /* 抽象方法，子类必须实现，负责设置daemon运行所必须的属性 */
    abstract function setPara();

    /* 抽象方法，子类必须实现，负责每条日志处理的过程 */
    abstract function deal();

    /* 收集文件MD5 */
    private function collectVerifyMd5() {
        $this->conf_md5 = md5_file($this->path_conf);
        $this->deal_md5 = md5_file($this->file_deal);
        $this->base_md5 = md5_file(__FILE__);
    }

    /* 校验三个文件是否修改，如修改，退出 */
    private function checkVerifyMd5() {
        if($this->conf_md5 != md5_file($this->path_conf)) exit("daemon conf file modified.\n");
        if($this->deal_md5 != md5_file($this->file_deal)) exit("deal file modified.\n");
        if($this->base_md5 != md5_file(__FILE__)) exit("base class file modified.\n");
    }

    /**
     * 设置进程健康状态
     *
     * @return	boolean
     */
    protected function setProcHealth() {
        $proc_path = sprintf($this->path_proc, $this->proc_code.'_'.md5($this->path_deal));

        $handle = fopen($proc_path, 'w');
        if(!$handle) {
            return false;
        }
        $content = date('Y-m-d H:i:s').'|'.$this->path_deal;
        if(!fwrite($handle, $content)) {
            return false;
        }
        fclose($handle);
    }

    /**
     * 进程上限限制
     *
     * @return	boolean
     */
    private function cProcessNum() {
        if($this->proc_code+1>$this->processnum) {
            //error_log('['.date('Y-m-d H:i:s').']'.$this->proc_code."|procnum:{$this->processnum}\n", 3, dirname(__FILE__).'/deal_track.log');
            return false;
        }

        $cmd = "ps -ef | grep '{$this->path_deal} {$this->proc_code}' | grep -v grep | grep -v '\/bin\/sh' | grep -v 'sh -c' | wc -l";
        $rcmd = @popen($cmd, 'r');
        $num = @fread($rcmd, 512);
        $num += 0;

        @pclose($rcmd);
        if($num > $this->processnum) {
            //error_log('['.date('Y-m-d H:i:s').']procnum:'."$num|{$this->processnum}|$cmd\n", 3, dirname(__FILE__).'/deal_track.log');
            return false;
        }
        return true;
    }

    /**
     * 跟踪日志
     * @param int $type
     * @param string $log
     * @return boolean
     */
    public function trackLog($type, $log, $log_path='') {
        if(empty($log_path)) $log_path = sprintf(LOG_TRACK_SAVE_PATH, date('Y-m-d'), $type);
        $log_dir = dirname($log_path);
        $isfirst = false;
        if(!is_dir($log_dir)) {
            mkdir($log_dir, 0775, true);
            chown($log_dir, 'nobody');
            chgrp($log_dir, 'nobody');
            $isfirst = true;
        }

        error_log(date('H:i:s').'|'.$log."\r\n", 3, $log_path);
        chown($log_path, 'nobody');
        chgrp($log_path, 'nobody');

        return true;
    }

}