channel = stream_socket_pair(STREAM_PF_UNIX, STREAM_SOCK_STREAM, STREAM_IPPROTO_IP); if($this->channel) { stream_set_blocking($this->channel[0], 0); $this->_readFds[0] = $this->channel[0]; } // 初始化优先队列(最大堆) $this->_scheduler = new \SplPriorityQueue(); $this->_scheduler->setExtractFlags(\SplPriorityQueue::EXTR_BOTH); } /** * 添加事件及处理函数 * @see Events\EventInterface::add() */ public function add($fd, $flag, $func, $args = null) { switch ($flag) { case self::EV_READ: $fd_key = (int)$fd; $this->_allEvents[$fd_key][$flag] = array($func, $fd); $this->_readFds[$fd_key] = $fd; break; case self::EV_WRITE: $fd_key = (int)$fd; $this->_allEvents[$fd_key][$flag] = array($func, $fd); $this->_writeFds[$fd_key] = $fd; break; case self::EV_SIGNAL: $fd_key = (int)$fd; $this->_signalEvents[$fd_key][$flag] = array($func, $fd); pcntl_signal($fd, array($this, 'signalHandler')); break; case self::EV_TIMER: case self::EV_TIMER_ONCE: // $fd 为 定时的时间间隔,单位为秒,支持小数,能精确到0.001秒 $run_time = microtime(true)+$fd; $this->_scheduler->insert($this->_timerId, -$run_time); $this->_task[$this->_timerId] = array($func, $args, $flag, $fd); $this->tick(); return $this->_timerId++; } return true; } /** * 信号处理函数 * @param int $signal */ public function signalHandler($signal) { call_user_func_array($this->_signalEvents[$signal][self::EV_SIGNAL][0], array($signal)); } /** * 删除某个描述符的某类事件的监听 * @see Events\EventInterface::del() */ public function del($fd ,$flag) { $fd_key = (int)$fd; switch ($flag) { case self::EV_READ: unset($this->_allEvents[$fd_key][$flag], $this->_readFds[$fd_key]); if(empty($this->_allEvents[$fd_key])) { unset($this->_allEvents[$fd_key]); } return true; case self::EV_WRITE: unset($this->_allEvents[$fd_key][$flag], $this->_writeFds[$fd_key]); if(empty($this->_allEvents[$fd_key])) { unset($this->_allEvents[$fd_key]); } return true; case self::EV_SIGNAL: unset($this->_signalEvents[$fd_key]); pcntl_signal($fd, SIG_IGN); break; case self::EV_TIMER: case self::EV_TIMER_ONCE; // $fd_key为要删除的定时器id,即timerId unset($this->_task[$fd_key]); return true; } return false;; } /** * 检查是否有可执行的定时任务,有的话执行 * @return void */ 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); if($time_now >= $next_run_time) { $this->_scheduler->extract(); // 如果任务不存在,则是对应的定时器已经删除 if(!isset($this->_task[$timer_id])) { continue; } // 任务数据[func, args, flag, timer_interval] $task_data = $this->_task[$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); } // 尝试执行任务 try { call_user_func_array($task_data[0], $task_data[1]); } catch(\Exception $e) { echo $e; } continue; } else { // 设定超时时间 $this->_selectTimeout = ($next_run_time - $time_now)*1000000; return; } } $this->_selectTimeout = 100000000; } /** * 删除所有定时器 * @return void */ public function clearAllTimer() { $this->_scheduler = new \SplPriorityQueue(); $this->_scheduler->setExtractFlags(\SplPriorityQueue::EXTR_BOTH); $this->_task = array(); } /** * 主循环 * @see Events\EventInterface::loop() */ public function loop() { $e = null; while (1) { // 如果有信号,尝试执行信号处理函数 pcntl_signal_dispatch(); $read = $this->_readFds; $write = $this->_writeFds; // 等待可读或者可写事件 @stream_select($read, $write, $e, 0, $this->_selectTimeout); // 这些描述符可读,执行对应描述符的读回调函数 if($read) { foreach($read as $fd) { $fd_key = (int) $fd; if(isset($this->_allEvents[$fd_key][self::EV_READ])) { call_user_func_array($this->_allEvents[$fd_key][self::EV_READ][0], array($this->_allEvents[$fd_key][self::EV_READ][1])); } } } // 这些描述符可写,执行对应描述符的写回调函数 if($write) { foreach($write as $fd) { $fd_key = (int) $fd; if(isset($this->_allEvents[$fd_key][self::EV_WRITE])) { call_user_func_array($this->_allEvents[$fd_key][self::EV_WRITE][0], array($this->_allEvents[$fd_key][self::EV_WRITE][1])); } } } // 尝试执行定时任务 if(!$this->_scheduler->isEmpty()) { $this->tick(); } } } }