【读源码】workerman中的2种Timer[B君笔记][博客]

noscript{position: fixed}【读源码】workerman中的2种Timerphpworkerman创建时间：2021-11-13 16:18:56

初始化

我们先看下Timer::init()在项目中用到的地方。主要区别是调用的地方和初始化的数据。

位置	主进程	子进程
参数	空	static::$globalEvent

/**
     * Init.
     *
     * @param EventInterface $event
     * @return void
     */
    public static function init($event = null)
    {
        if ($event) {
            self::$_event = $event;
            return;
        }
        if (\function_exists('pcntl_signal')) {
            \pcntl_signal(\SIGALRM, array('\Workerman\Lib\Timer', 'signalHandle'), false);
        }
    }

主进程的Timer

从源码可以看出，这个Timer 基于进程alarm闹钟信号的定时器。为SIGALRM信号安装了一个信号处理器\Workerman\Lib\Timer::signalHandle。

\Workerman\Lib\Timer::signalHandle

作为信号处理器，每次收到信号SIGALRM，都会执行一次。

   /**
     * ALARM signal handler.
     *
     * @return void
     */
    public static function signalHandle()
    {
        if (!self::$_event) {
//1秒数后向进程发送一个SIGALRM信号
            \pcntl_alarm(1);
//触发一次任务检查
            self::tick();
        }
    }

\Workerman\Lib\Timer::tick();

    /**
     * Tick.
     *
     * @return void
     */
    public static function tick()
    {
	//没有任务，关闭闹钟
        if (empty(self::$_tasks)) {
            \pcntl_alarm(0);
            return;
        }
	// 处理任务
        $time_now = \time();
        foreach (self::$_tasks as $run_time => $task_data) {
		//满足时间要求
            if ($time_now >= $run_time) {
                foreach ($task_data as $index => $one_task) {
                    $task_func     = $one_task[0];
                    $task_args     = $one_task[1];
                    $persistent    = $one_task[2];
                    $time_interval = $one_task[3];
                    try {
			//执行任务
                        \call_user_func_array($task_func, $task_args);
                    } catch (\Exception $e) {
                        Worker::safeEcho($e);
                    }
		    //不是一次性任务，设定好下一次时间，并添加任务
                    if($persistent && !empty(self::$_status[$index])) {
                        $new_run_time = \time() + $time_interval;
                        if(!isset(self::$_tasks[$new_run_time])) self::$_tasks[$new_run_time] = array();
                        self::$_tasks[$new_run_time][$index] = array($task_func, (array)$task_args, $persistent, $time_interval);
                    }
                }
                unset(self::$_tasks[$run_time]);
            }
        }
    }

monitorWorkersForLinux

最后循环监听进程信号

protected static function monitorWorkersForLinux()
    {
        static::$_status = static::STATUS_RUNNING;
        while (1) {
            // Calls signal handlers for pending signals.
            \pcntl_signal_dispatch();
            // Suspends execution of the current process until a child has exited, or until a signal is delivered
            $status = 0;
            $pid    = \pcntl_wait($status, \WUNTRACED);
            // Calls signal handlers for pending signals again.
            \pcntl_signal_dispatch();
	//....
	}
}

示例

<?php
use \Workerman\Worker;

require './vendor/autoload.php';
// WebServer
$worker = new Worker("tcp://0.0.0.0:2000");
// WebServer数量
$worker->count = 1;

\Workerman\Timer::add(1,function(){
   echo time().PHP_EOL;
});
// 如果不是在根目录启动，则运行runAll方法
if(!defined('GLOBAL_START'))
{
    Worker::runAll();
}

运行结果

总结

主进程定时器的运行流程如下：

初始化定时器，设定信号处理器
Timer::add 时，设定闹钟并触发检查
没有任务时移除闹钟
有任务时每秒检查符合要求的任务并执行，完成后视情况添加重复任务
Timer清除时，清除闹钟

注意的：

由于基于信号，时间精度只到秒
下次任务时间是基于上次任务的结束时间
耗时的定时任务会阻塞进程阻塞
该类型只在linux上的主进程上可以使用

尴尬

这个类型的Timer只适用于非守护进程模式，也就是只在debug模式下生效。原因是，主进程会被退出了，详情可见【读源码】workerman中如何实现守护进程

子进程的Timer

子进程的Timer机制时基于even类型，如：libeven,event,swoole或者react，默认的是：\Workerman\Events\Select。此处以它为例。### win没有多进程
在win中workerman是以单进程形式运行，在第二次初始化Timer时，其实时覆盖了主进程的Timer。也就是说，在win中，Timer只有一个类型。

我们以linux为例。

子进程中的Timer都是在被fork后进行初始化，并且会出入参数 static::$globalEvent

static::$globalEvent

	// Create a global event loop.
        if (!static::$globalEvent) {
            $event_loop_class = static::getEventLoopName();
            static::$globalEvent = new $event_loop_class;
            $this->resumeAccept();
        }
        Timer::init(static::$globalEvent);

getEventLoopName()


    /**
     * Available event loops.
     *
     * @var array
     */
    protected static $_availableEventLoops = array(
        'event'    => '\Workerman\Events\Event',
        'libevent' => '\Workerman\Events\Libevent'
    );

    /**
     * Get event loop name.
     *
     * @return string
     */
    protected static function getEventLoopName()
    {
        if (static::$eventLoopClass) {
            return static::$eventLoopClass;
        }

        if (!\class_exists('\Swoole\Event', false)) {
            unset(static::$_availableEventLoops['swoole']);
        }

        $loop_name = '';
        foreach (static::$_availableEventLoops as $name=>$class) {
            if (\extension_loaded($name)) {
                $loop_name = $name;
                break;
            }
        }

        if ($loop_name) {
            if (\interface_exists('\React\EventLoop\LoopInterface')) {
                switch ($loop_name) {
                    case 'libevent':
                        static::$eventLoopClass = '\Workerman\Events\React\ExtLibEventLoop';
                        break;
                    case 'event':
                        static::$eventLoopClass = '\Workerman\Events\React\ExtEventLoop';
                        break;
                    default :
                        static::$eventLoopClass = '\Workerman\Events\React\StreamSelectLoop';
                        break;
                }
            } else {
                static::$eventLoopClass = static::$_availableEventLoops[$loop_name];
            }
        } else {
            static::$eventLoopClass = \interface_exists('\React\EventLoop\LoopInterface') ? '\Workerman\Events\React\StreamSelectLoop' : '\Workerman\Events\Select';
        }
        return static::$eventLoopClass;
    }

可以看出static::$globalEvent的优先级如下：

event
1. \Workerman\Events\React\ExtEventLoop
2. \Workerman\Events\Event
libevent
1. \Workerman\Events\React\ExtLibEventLoop
2. \Workerman\Events\Libevent
\Workerman\Events\React\StreamSelectLoop
\Workerman\Events\Select

\Workerman\Events\Select

终于到我们今天的主角了，我们看看这里的Timer是怎么实现的

Timer转发

可以看出Timer中的命令都被转发到了 static::$globalEvent

<?php
	//Timer::add
	 if (self::$_event) {
            return self::$_event->add($time_interval,
                $persistent ? EventInterface::EV_TIMER : EventInterface::EV_TIMER_ONCE, $func, $args);
        }

	// Timer::del
	   if (self::$_event) {
            return self::$_event->del($timer_id, EventInterface::EV_TIMER);
        }

	// Timer::delAll
	 if (self::$_event) {
            self::$_event->clearAllTimer();
        }

Select 之 SplPriorityQueue

在\Workerman\Events\Select初始化时，同时会初始化一个SplPriorityQueue，用于存放定时任务。
SplPriorityQueue说php中内置的优先级队列，以Heap数据结构实现，默认为MaxHeap模式，即priority越大越优先出队。


    /**
     * Construct.
     */
    public function __construct()
    {
        // Init SplPriorityQueue.
        $this->_scheduler = new \SplPriorityQueue();
        $this->_scheduler->setExtractFlags(\SplPriorityQueue::EXTR_BOTH);
    }

add

    public function add($fd, $flag, $func, $args = array())
    {
        switch ($flag) {
            //...
            case self::EV_TIMER:
            case self::EV_TIMER_ONCE:
		//定时器编号
                $timer_id = $this->_timerId++;
                $run_time = \microtime(true) + $fd;
		//以时间负值为优先级，插入到队列中
                $this->_scheduler->insert($timer_id, -$run_time);
		//标记对应定时器的回调
                $this->_eventTimer[$timer_id] = array($func, (array)$args, $flag, $fd);
                $select_timeout = ($run_time - \microtime(true)) * 1000000;
                if( $this->_selectTimeout > $select_timeout ){ 
                    $this->_selectTimeout = $select_timeout;   
                }  
		//返回定时器编号
                return $timer_id;
        }

        return true;
    }

del

    public function del($fd, $flag)
    {
        $fd_key = (int)$fd;
        switch ($flag) {
            //...
            case self::EV_TIMER:
            case self::EV_TIMER_ONCE;
		//移除对应定时器的回
                unset($this->_eventTimer[$fd_key]);
                return true;
        }
        return false;
    }

delAll=>clearAllTimer

    public function clearAllTimer()
    {
	//直接重新初始化的队列和存放定时器回调数组
        $this->_scheduler = new \SplPriorityQueue();
        $this->_scheduler->setExtractFlags(\SplPriorityQueue::EXTR_BOTH);
        $this->_eventTimer = array();
    }

event::tick

protected function tick()
    {
        while (!$this->_scheduler->isEmpty()) {
 	   
            $scheduler_data       = $this->_scheduler->top();
            $timer_id             = $scheduler_data['data'];
            $next_run_time        = -$scheduler_data['priority'];
            $time_now             = \microtime(true);
            $this->_selectTimeout = ($next_run_time - $time_now) * 1000000;
            if ($this->_selectTimeout <= 0) {
		//Extract出队
                $this->_scheduler->extract();
		//没有对应回调，直接返回
                if (!isset($this->_eventTimer[$timer_id])) {
                    continue;
                }

                // [func, args, flag, timer_interval]
                $task_data = $this->_eventTimer[$timer_id];
                if ($task_data[2] === self::EV_TIMER) {
                    $next_run_time = $time_now + $task_data[3];
		    //插入下一次任务
                    $this->_scheduler->insert($timer_id, -$next_run_time);
                }
		//执行任务
                \call_user_func_array($task_data[0], $task_data[1]);
                if (isset($this->_eventTimer[$timer_id]) && $task_data[2] === self::EV_TIMER_ONCE) {
		   //单次任务的移除下一次任务
                    $this->del($timer_id, self::EV_TIMER_ONCE);
                }
                continue;
            }
	    //没有可以执行的任务跳出循环
            return;
        }
        $this->_selectTimeout = 100000000;
    }

event::loop

相比于主进程Timer的闹钟信号，子进程的通过event->loop循环进行处理。每个子进程最后都是进入一个循环，在这个循环中监听网络事件和定时任务。正如文档中的：如果任务很重（特别是涉及到网络IO的任务），可能会导致该进程阻塞，暂时无法处理其它业务。

    public function loop()
    {
        while (1) {
            if(\DIRECTORY_SEPARATOR === '/') {
                // Calls signal handlers for pending signals
                \pcntl_signal_dispatch();
            }

            $read  = $this->_readFds;
            $write = $this->_writeFds;
            $except = $this->_exceptFds;

            if ($read || $write || $except) {
                // Waiting read/write/signal/timeout events.
                try {
                    $ret = @stream_select($read, $write, $except, 0, $this->_selectTimeout);
                } catch (\Exception $e) {} catch (\Error $e) {}

            } else {
		// 单纯的定时任务会进行休眠
                usleep($this->_selectTimeout);
                $ret = false;
            }

	    //  有任务，进去检查一下
            if (!$this->_scheduler->isEmpty()) {
                $this->tick();
            }

            if (!$ret) {
                continue;
            }
	    //...
        }
    }

总结

Timer在\Workerman\Events\Select中的实现主基于SplPriorityQueue优先队列
Timer在循环中并无其他限制，所以可以精确到毫秒
Timer没错执行都是在网络事件的回调之前，也许可以做点事情

其他的在文档中已经写得很详细了，可以移步Timer定时器类查看

函数参考

pcntl_signal

说明

pcntl_signal(int $signo, callback $handler, bool $restart_syscalls = true): bool

函数pcntl_signal()为signo指定的信号安装一个新的信号处理器。

参数
- signo
  信号编号。
- handler
  信号处理器可以是用户创建的函数或方法的名字，也可以是系统常量 SIG_IGN（译注：忽略信号处理程序）或SIG_DFL（默认信号处理程序）.
  
  注意:
  注意当你使用一个对象方法的时候，该对象的引用计数回增加使得它在你改变为其他处理或脚本结束之前是持久存在的。
- restart_syscalls
  指定当信号到达时系统调用重启是否可用。（译注：经查资料，此参数意为系统调用被信号打断时，系统调用是否从开始处重新开始，但根据http://bugs.php.net/bug.php?id=52121，此参数存在bug无效。）
返回值
成功时返回 true，或者在失败时返回 false。

pcntl_alarm

说明

pcntl_alarm(int $seconds): int

创建一个计时器，在指定的秒数后向进程发送一个SIGALRM信号。每次对 pcntl_alarm()的调用都会取消之前设置的alarm信号。

参数
- seconds
  等待的秒数。如果seconds设置为0,将不会创建alarm信号。
返回
返回上次alarm调度（离alarm信号发送）剩余的秒数，或者之前没有alarm调度（译注：或者之前调度已完成）时返回0。

参考

phptimer
项目介绍：phptimer现在是一个定时器项目，参考了workerman里面的源码，剥离了网络和多进程，保留了毫秒定时器，基于stream_select和event的阻塞超时实现。