php Workerman-Lib-Timer类(方法)实例源码

/**
  * 初始化一些环境变量
  *
  * @return void
  */
 public static function init()
 {
     // 如果没设置$pidFile，则生成默认值
     if (empty(self::$pidFile)) {
         $backtrace = debug_backtrace();
         self::$_startFile = $backtrace[count($backtrace) - 1]['file'];
         self::$pidFile = sys_get_temp_dir() . "/workerman." . str_replace('/', '_', self::$_startFile) . ".pid";
     }
     // 没有设置日志文件，则生成一个默认值
     if (empty(self::$logFile)) {
         self::$logFile = __DIR__ . '/../server.log';
     }
     // 标记状态为启动中
     self::$_status = self::STATUS_STARTING;
     // 启动时间戳
     self::$_globalStatistics['start_timestamp'] = time();
     Base::getLog()->debug(__METHOD__ . ' worker start timestamp', ['time' => self::$_globalStatistics['start_timestamp']]);
     // 设置status文件位置
     self::$statusFile = sys_get_temp_dir() . '/workerman.status';
     Base::getLog()->debug(__METHOD__ . ' set status file', ['file' => self::$statusFile]);
     // 尝试设置进程名称（需要php>=5.5或者安装了proctitle扩展）
     self::setProcessTitle('WorkerMan: master process  start_file=' . self::$_startFile);
     // 初始化ID
     self::initId();
     // 初始化定时器
     Timer::init();
 }

/**
  * 设置消息回调
  * @return void
  */
 public function onMessage($connection, $data)
 {
     // 删除定时器
     Timer::del($connection->timeout_timerid);
     $data = json_decode($data, true);
     $event = $data['event'];
     // 开始验证
     switch ($event) {
         // 是geteway连接
         case 'gateway_connect':
             if (empty($data['address'])) {
                 echo "address not found\n";
                 return $connection->close();
             }
             $this->_gatewayConnections[$connection->id] = $data['address'];
             $this->broadcastAddresses();
             break;
             // 是worker连接
         // 是worker连接
         case 'worker_connect':
             $this->_workerConnections[$connection->id] = $connection;
             $this->broadcastAddresses($connection);
             break;
         case 'ping':
             break;
         default:
             echo "unknown event {$event}\n";
             $connection->close();
     }
 }

public static function clearTimer()
 {
     if (self::$_reconnectTimer) {
         Timer::del(self::$_reconnectTimer);
         self::$_reconnectTimer = null;
     }
 }

public function scheduleRespawn($delay)
 {
     Timer::add($delay / 1000, function ($self) {
         if ($self->respawnCallback) {
             call_user_func($self->respawnCallback);
         }
     }, array($this), false);
 }

/**
  * 当进程启动时一些初始化工作
  * @return void
  */
 protected function onWorkerStart()
 {
     Timer::add(1, array($this, 'checkGatewayConnections'));
     $this->checkGatewayConnections();
     \GatewayWorker\Lib\Gateway::setBusinessWorker($this);
     if ($this->_onWorkerStart) {
         call_user_func($this->_onWorkerStart, $this);
     }
 }

/**
  * 牌局玩家进度广播
  */
 public static function doApi($player, $data, &$re, $client_id)
 {
     $uid = $player->uid;
     $table = TableDao::getTable($player->tableId);
     if (!$table) {
         return 1;
     }
     if (!Timer::isExistTimer($table->blinkTimeOut)) {
         Gateway::bindUid($client_id, $uid);
         $table->blinkTimeOut = Timer::add(Constants::TABLE_INIT_CHECK_TIME, array($table, 'checkTime'));
         TableDao::addTable($table->tableId, $table);
     }
     if (!isset($table->playerStatus[$uid])) {
         $table->addUid($uid);
         GameDao::addInGamePlayer($uid);
         TableDao::addTable($table->tableId, $table);
     }
     if ($data['st'] == 1) {
         if (!in_array($uid, $table->readyUids)) {
             $table->readyUids[] = $uid;
             TableDao::addTable($table->tableId, $table);
         }
         if (count($table->readyUids) >= 3) {
             $table->recordTime = time();
             TableDao::addTable($table->tableId, $table);
             $re['uid'] = -1;
             Gateway::sendToUid($table->uids, json_encode($re));
         }
     }
     if ($data['addVal'] != -1) {
         $uids = $table->uids;
         $re['uid'] = $uid;
         foreach ($uids as $_uid) {
             if ($_uid == $uid) {
                 continue;
             }
             $re['addVal'] = $data['addVal'];
             $re['oldVal'] = $data['oldVal'];
             Gateway::sendToUid($_uid, json_encode($re));
         }
     }
     return 1;
 }

/**
  * Display a listing of the resource.
  *
  * @return Response
  */
 public function index()
 {
     require_once '/vendor/workerman/workerman/Autoloader.php';
     $http_worker = new Worker("websocket://0.0.0.0:2345");
     $http_worker->count = 4;
     $http_worker->onConnect = function ($connection) {
         $connection->send('id' . $connection->id);
     };
     $http_worker->onWorkerStart = function ($worker) {
         // 定时，每10秒一次
         \Workerman\Lib\Timer::add(2, function () use($worker) {
             // 遍历当前进程所有的客户端连接，发送当前服务器的时间
             foreach ($worker->connections as $connection) {
                 $connection->send(time());
             }
         });
     };
     $http_worker->onMessage = function ($connection, $data) {
         $connection->send('hello world' . $data);
     };
     worker::runAll();
 }

public function initRoaming($mob)
 {
     Timer::add(0.5, array($this, 'initRoamingCallback'));
 }

/**
  * 当与Gateway的连接断开时触发
  * @param TcpConnection $connection
  * @return  void
  */
 public function onGatewayClose($connection)
 {
     $addr = $connection->remoteAddress;
     unset($this->gatewayConnections[$addr], $this->_connectingGatewayAddresses[$addr]);
     if (isset($this->_gatewayAddresses[$addr]) && !isset($this->_waitingConnectGatewayAddresses[$addr])) {
         Timer::add(1, array($this, 'tryToConnectGateway'), array($addr), false);
         $this->_waitingConnectGatewayAddresses[$addr] = $addr;
     }
 }

public function onRegisterConnectionClose()
 {
     Timer::add(1, array($this, 'registerAddress'), null, false);
 }

<?php

use Workerman\Worker;
use Workerman\Lib\Timer;
// composer autoload
include __DIR__ . '/../vendor/autoload.php';
$channel_server = new Channel\Server();
$worker = new Worker();
$worker->onWorkerStart = function () {
    Channel\Client::on('test event', function ($event_data) {
        echo 'test event triggered event_data :';
        var_dump($event_data);
    });
    Timer::add(5, function () {
        Channel\Client::publish('test event', 'some data');
    });
};
Worker::runAll();

\Workerman\Lib\Timer::add(0.5, function () use($consumer) {
     if (extension_loaded('sysvmsg')) {
         // 循环取数据
         while (1) {
             $desiredmsgtype = 1;
             $msgtype = 0;
             $message = '';
             $maxsize = 65535;
             // 从队列中获取消息 @see http://php.net/manual/zh/function.msg-receive.php
             @msg_receive($consumer->queue, $desiredmsgtype, $msgtype, $maxsize, $message, true, MSG_IPC_NOWAIT);
             if (!$message) {
                 return;
             }
             // 假设消息数据为json，格式类似{"class":"class_name", "method":"method_name", "args":[]}
             $message = json_decode($message, true);
             // 格式如果是正确的，则尝试执行对应的类方法
             if (isset($message['class']) && isset($message['method']) && isset($message['args'])) {
                 // 要调用的类名，加上Consumer命名空间
                 $class_name = "\\Consumer\\" . $message['class'];
                 // 要调用的方法名
                 $method = $message['method'];
                 // 调用参数，是个数组
                 $args = (array) $message['args'];
                 // 类存在则尝试执行
                 if (class_exists($class_name)) {
                     $class = new $class_name();
                     $callback = array($class, $method);
                     if (is_callable($callback)) {
                         call_user_func_array($callback, $args);
                     } else {
                         echo "{$class_name}::{$method} not exist\n";
                     }
                 } else {
                     echo "{$class_name} not exist\n";
                 }
             } else {
                 echo "unknow message\n";
             }
         }
     }
 });

/**
  * 当Gateway启动的时候触发的回调函数
  * @return void
  */
 public function onWorkerStart()
 {
     // 分配一个内部通讯端口
     $this->lanPort = function_exists('posix_getppid') ? $this->startPort - posix_getppid() + posix_getpid() : $this->startPort;
     if ($this->lanPort < 0 || $this->lanPort >= 65535) {
         $this->lanPort = rand($this->startPort, 65535);
     }
     // 如果有设置心跳，则定时执行
     if ($this->pingInterval > 0) {
         $timer_interval = $this->pingNotResponseLimit > 0 ? $this->pingInterval / 2 : $this->pingInterval;
         Timer::add($timer_interval, array($this, 'ping'));
     }
     if (!class_exists('\\Protocols\\GatewayProtocol')) {
         class_alias('\\GatewayWorker\\Protocols\\GatewayProtocol', 'Protocols\\GatewayProtocol');
     }
     // 初始化gateway内部的监听，用于监听worker的连接已经连接上发来的数据
     $this->_innerTcpWorker = new Worker("GatewayProtocol://{$this->lanIp}:{$this->lanPort}");
     $this->_innerTcpWorker->listen();
     $this->_innerUdpWorker = new Worker("GatewayProtocol://{$this->lanIp}:{$this->lanPort}");
     $this->_innerUdpWorker->transport = 'udp';
     $this->_innerUdpWorker->listen();
     // 重新设置自动加载根目录
     Autoloader::setRootPath($this->_appInitPath);
     // 设置内部监听的相关回调
     $this->_innerTcpWorker->onMessage = array($this, 'onWorkerMessage');
     $this->_innerUdpWorker->onMessage = array($this, 'onWorkerMessage');
     $this->_innerTcpWorker->onConnect = array($this, 'onWorkerConnect');
     $this->_innerTcpWorker->onClose = array($this, 'onWorkerClose');
     // 注册gateway的内部通讯地址，worker去连这个地址，以便gateway与worker之间建立起TCP长连接
     if (!$this->registerAddress()) {
         $this->log('registerAddress fail and exit');
         Worker::stopAll();
     }
     if ($this->_onWorkerStart) {
         call_user_func($this->_onWorkerStart, $this);
     }
 }

$connection->close();
        }
        // onWebSocketConnect 里面$_GET $_SERVER是可用的
        // var_dump($_GET, $_SERVER);
    };
}; 
*/
$webserver = new WebServer('http://0.0.0.0:80');
$webserver->addRoot('120.25.163.9', '/workerman/Applications/huicheng/Web');
$webserver->count = 1;
$webserver->onWorkerStart = function ($webserver) {
    //初始化多客服工号对应的昵称
    redisData::Set('HCS1@hc-information', '小汇');
    redisData::Set('HCS2@hc-information', '小承');
    redisData::Set('HCT1@hc-information', '小信');
    redisData::Set('HCT2@hc-information', '小息');
    //初始化定时间隔
    $time_interval = 7000;
    //初始化access_token
    transToWxServer::getaccess_token();
    transToWxServer::getjsapi_ticket();
    //定时获取access_token
    \Workerman\Lib\Timer::add($time_interval, function () {
        transToWxServer::getaccess_token();
        transToWxServer::getjsapi_ticket();
    });
};
// 如果不是在根目录启动，则运行runAll方法
if (!defined('GLOBAL_START')) {
    Worker::runAll();
}

/**
  * Run worker instance.
  * @return void
  */
 public function run()
 {
     //Update process state.
     self::$_status = self::STATUS_RUNNING;
     // Eegister shutdown function for checking errors.
     register_shutdown_function(array("\\Workerman\\Worker", 'checkErrors'));
     // Set autoload root path.
     Autoloader::setRootPath($this->_autoloadRootPath);
     // Create a global event loop.
     if (!self::$globalEvent) {
         $eventLoopClass = "\\Workerman\\Events\\" . ucfirst(self::getEventLoopName());
         self::$globalEvent = new $eventLoopClass();
         // Register a listener to be notified when server socket is ready to read.
         if ($this->_socketName) {
             if ($this->transport !== 'udp') {
                 self::$globalEvent->add($this->_mainSocket, EventInterface::EV_READ, array($this, 'acceptConnection'));
             } else {
                 self::$globalEvent->add($this->_mainSocket, EventInterface::EV_READ, array($this, 'acceptUdpConnection'));
             }
         }
     }
     // Reinstall signal.
     self::reinstallSignal();
     // Init Timer.
     Timer::init(self::$globalEvent);
     // Try to emit onWorkerStart callback.
     if ($this->onWorkerStart) {
         try {
             call_user_func($this->onWorkerStart, $this);
         } catch (\Exception $e) {
             echo $e;
             exit(250);
         }
     }
     // Main loop.
     self::$globalEvent->loop();
 }

<?php

use Workerman\Worker;
use Workerman\Lib\Timer;
// composer autoload
include __DIR__ . '/../vendor/autoload.php';
$channel_server = new Channel\Server();
$worker = new Worker();
$worker->onWorkerStart = function () {
    Channel\Client::$onMessage = function ($channel, $data) {
        var_dump($channel, $data);
    };
    Channel\Client::subscribe('abc');
    Timer::add(5, function () {
        Channel\Client::publish('abc', array('efg'));
    });
};
Worker::runAll();

/**
  * 当用户断开连接时触发
  * @param  int $client_id
  * @return void
  */
 public static function onClose($client_id)
 {
     \Workerman\Lib\Timer::del(self::$redisConnection->get($client_id));
     self::$redisConnection->del($client_id);
     // 更新MYSQL数据库
     self::$connectHC->query("UPDATE `HC` SET `clientid` = 0, `lastouttime` = CURRENT_TIMESTAMP() WHERE clientid='{$client_id}'");
 }

public function returnToSpawningPosition($waitDuration)
 {
     $delay = $waitDuration ? $waitDuration : 4000;
     $this->clearTarget();
     $this->returnTimeout = Timer::add($delay / 1000, array($this, 'timeoutCallback'), array(), false);
 }

$sender_io->to($to)->emit('new_msg', @$_POST['content']);
                    // 否则向所有uid推送数据
                } else {
                    $sender_io->emit('new_msg', @$_POST['content']);
                }
                // http接口返回ok
                return $http_connection->send('ok');
        }
        return $http_connection->send('fail');
    };
    // 执行监听
    $inner_http_worker->listen();
    // 一个定时器，定时向所有uid推送当前uid在线数及在线页面数
    Timer::add(1, function () {
        global $uidConnectionMap, $sender_io, $last_online_count, $last_online_page_count;
        $online_count_now = count($uidConnectionMap);
        $online_page_count_now = array_sum($uidConnectionMap);
        // 只有在客户端在线数变化了才广播，减少不必要的客户端通讯
        if ($last_online_count != $online_count_now || $last_online_page_count != $online_page_count_now) {
            $sender_io->emit('update_online_count', "当前<b>{$online_count_now}</b>人在线，共打开<b>{$online_page_count_now}</b>个页面");
            $last_online_count = $online_count_now;
            $last_online_page_count = $online_page_count_now;
        }
    });
});
// 启动一个webserver，用于吐html css js，方便展示
// 这个webserver服务不是必须的，可以将这些html css js文件放到你的项目下用nginx或者apache跑
$web = new WebServer('http://0.0.0.0:2123');
$web->addRoot('localhost', __DIR__ . '/web');
// 运行所有的服务
Worker::runAll();

/**
  * Websocket handshake.
  *
  * @param string                              $buffer
  * @param \Workerman\Connection\TcpConnection $connection
  * @return int
  */
 public static function dealHandshake($buffer, $connection)
 {
     $pos = strpos($buffer, "\r\n\r\n");
     if ($pos) {
         // handshake complete
         $connection->handshakeStep = 2;
         $handshake_response_length = $pos + 4;
         // Try to emit onWebSocketConnect callback.
         if (isset($connection->onWebSocketConnect)) {
             try {
                 call_user_func($connection->onWebSocketConnect, $connection, substr($buffer, 0, $handshake_response_length));
             } catch (\Exception $e) {
                 Worker::log($e);
                 exit(250);
             } catch (\Error $e) {
                 Worker::log($e);
                 exit(250);
             }
         }
         // Headbeat.
         if (!empty($connection->websocketPingInterval)) {
             $connection->websocketPingTimer = Timer::add($connection->websocketPingInterval, function () use($connection) {
                 if (false === $connection->send(pack('H*', '8900'), true)) {
                     Timer::del($connection->websocketPingTimer);
                     $connection->websocketPingTimer = null;
                 }
             });
         }
         $connection->consumeRecvBuffer($handshake_response_length);
         if (!empty($connection->tmpWebsocketData)) {
             $connection->send($connection->tmpWebsocketData, true);
             $connection->tmpWebsocketData = '';
         }
         if (strlen($buffer) > $handshake_response_length) {
             return self::input(substr($buffer, $handshake_response_length), $connection);
         }
     }
     return 0;
 }