public function send($request, $future, $context, $conn)
 {
     $self = $this;
     $timeout = $context->timeout;
     if ($timeout > 0) {
         $conn->timeoutId = swoole_timer_after($timeout, function () use($self, $future, $conn) {
             $future->reject(new TimeoutException('timeout'));
             if ($conn->isConnected()) {
                 $conn->close();
             }
         });
     }
     $conn->onreceive = function ($conn, $data) use($self, $future) {
         $self->clean($conn);
         $self->sendNext($conn);
         $future->resolve($data);
     };
     $conn->onclose = function ($conn) use($self, $future) {
         $self->clean($conn);
         if ($conn->errCode !== 0) {
             $future->reject(new Exception(socket_strerror($conn->errCode)));
         } else {
             $future->reject(new Exception('The server is closed.'));
         }
         $self->size--;
     };
     $header = pack('N', strlen($request));
     $conn->send($header);
     $conn->send($request);
 }

function timeout($tm)
{
    echo time() . ": Timeout #{$tm}\n";
    if ($tm == 5) {
        swoole_timer_after(3000, 'timeout', 7);
    }
}

 public function send(callable $callback)
 {
     $client = new \swoole_client(SWOOLE_SOCK_TCP, SWOOLE_SOCK_ASYNC);
     $client->on("connect", function ($cli) {
         $cli->send($this->data);
     });
     $client->on('close', function ($cli) {
     });
     $client->on('error', function ($cli) use($callback) {
         $cli->close();
         call_user_func_array($callback, array('r' => 1, 'key' => $this->key, 'calltime' => $this->calltime, 'error_msg' => 'conncet error'));
     });
     $client->on("receive", function ($cli, $data) use($callback) {
         $this->calltime = microtime(true) - $this->calltime;
         $cli->close();
         call_user_func_array($callback, array('r' => 0, 'key' => $this->key, 'calltime' => $this->calltime, 'data' => $data));
     });
     if ($client->connect($this->ip, $this->port, $this->timeout, 1)) {
         $this->calltime = microtime(true);
         if (floatval($this->timeout) > 0) {
             $this->timer = swoole_timer_after(floatval($this->timeout) * 1000, function () use($client, $callback) {
                 $client->close();
                 \SysLog::error(__METHOD__ . " TIMEOUT ", __CLASS__);
                 $this->calltime = microtime(true) - $this->calltime;
                 call_user_func_array($callback, array('r' => 2, 'key' => '', 'calltime' => $this->calltime, 'error_msg' => 'timeout'));
             });
         }
     }
 }

 public function sendData(callable $callback)
 {
     $client = new swoole_client(SWOOLE_SOCK_TCP, SWOOLE_SOCK_ASYNC);
     $client->on("connect", function ($cli) {
         $cli->send($this->data);
     });
     $client->on('close', function ($cli) {
     });
     $client->on('error', function ($cli) use($callback) {
         $cli->close();
         call_user_func_array($callback, array('r' => 1, 'key' => $this->key, 'error_msg' => 'conncet error'));
     });
     $client->on("receive", function ($cli, $data) use($callback) {
         $cli->close();
         call_user_func_array($callback, array('r' => 0, 'key' => $this->key, 'data' => $data));
     });
     if ($client->connect($this->ip, $this->port, $this->timeout)) {
         if (intval($this->timeout) > 0) {
             swoole_timer_after(intval($this->timeout) * 1000, function () use($client, $callback) {
                 if ($client->isConnected()) {
                     $client->close();
                     call_user_func_array($callback, array('r' => 2, 'key' => '', 'error_msg' => 'timeout'));
                 }
             });
         }
     }
 }

 public function sendData(callable $callback)
 {
     $client = new swoole_client(SWOOLE_SOCK_UDP, SWOOLE_SOCK_ASYNC);
     $client->on("connect", function ($cli) {
         $this->isConnect = true;
         $cli->send($this->data);
     });
     $client->on('close', function ($cli) {
         $this->isConnect = false;
     });
     $client->on('error', function ($cli) use($callback) {
         $this->isConnect = false;
         $cli->close();
         call_user_func_array($callback, array('r' => 1, 'key' => $this->key, 'error_msg' => 'conncet error'));
     });
     $client->on("receive", function ($cli, $data) use($callback) {
         $this->isConnect = false;
         $cli->close();
         call_user_func_array($callback, array('r' => 0, 'key' => $this->key, 'data' => $data));
     });
     if ($client->connect($this->ip, $this->port, $this->timeout)) {
         if (intval($this->timeout) > 0) {
             swoole_timer_after(intval($this->timeout) * 1000, function () use($client, $callback) {
                 if ($this->isConnect) {
                     //error_log(__METHOD__." client ===== ".print_r($client,true),3,'/tmp/client.log');
                     $client->close();
                     call_user_func_array($callback, array('r' => 2, 'key' => '', 'error_msg' => 'timeout'));
                 }
             });
         }
     }
 }

 /**
  * [init 启动定时器]
  * @return [type] [description]
  */
 public static function init()
 {
     if (!self::$isOnTimer) {
         swoole_timer_after(1000 * self::LOOPTIME, function () {
             //循环数组，踢出超时情况
             self::loop();
             self::$isOnTimer = false;
         });
         self::$isOnTimer = true;
     }
 }

 public function __construct()
 {
     $fp = stream_socket_client("tcp://127.0.0.1:9504", $code, $msg, 3);
     $http_request = "GET /index.html HTTP/1.1\r\n\r\n";
     fwrite($fp, $http_request);
     swoole_event_add($fp, function ($fp) {
         echo fread($fp, 8192);
         swoole_event_del($fp);
         fclose($fp);
     });
     swoole_timer_after(2000, function () {
         echo "2000ms timeout\n";
     });
     swoole_timer_tick(1000, function () {
         echo "1000ms interval\n";
     });
 }

 function connect($host, $port)
 {
     if (empty($this->host)) {
         $this->host = $host;
         $this->port = $port;
     }
     $client = new swoole_client(SWOOLE_SOCK_TCP, SWOOLE_SOCK_ASYNC);
     $client->on("connect", [$this, 'onConnect']);
     $client->on("receive", function (swoole_client $cli, $data) {
         $cli->send("HELLO");
         echo "recv from server: {$data}\n";
         usleep(100000);
     });
     $client->on("error", [$this, 'onError']);
     $client->on("close", [$this, 'onClose']);
     $client->connect($host, $port);
     $this->timer = swoole_timer_after($this->timeout, [$this, 'onConnectTimeout']);
     $this->swoole_client = $client;
 }

 /**
  * @param $serverPid
  * @throws NotFound
  */
 function __construct($serverPid)
 {
     $this->pid = $serverPid;
     if (posix_kill($serverPid, 0) === false) {
         throw new NotFound("Process#{$serverPid} not found.");
     }
     $this->inotify = inotify_init();
     $this->events = IN_MODIFY | IN_DELETE | IN_CREATE | IN_MOVE;
     swoole_event_add($this->inotify, function ($ifd) {
         $events = inotify_read($this->inotify);
         if (!$events) {
             return;
         }
         var_dump($events);
         foreach ($events as $ev) {
             if ($ev['mask'] == IN_IGNORED) {
                 continue;
             } else {
                 if ($ev['mask'] == IN_CREATE or $ev['mask'] == IN_DELETE or $ev['mask'] == IN_MODIFY or $ev['mask'] == IN_MOVED_TO or $ev['mask'] == IN_MOVED_FROM) {
                     $fileType = strstr($ev['name'], '.');
                     //非重启类型
                     if (!isset($this->reloadFileTypes[$fileType])) {
                         continue;
                     }
                 }
             }
             //正在reload，不再接受任何事件，冻结10秒
             if (!$this->reloading) {
                 $this->putLog("after 10 seconds reload the server");
                 //有事件发生了，进行重启
                 swoole_timer_after($this->afterNSeconds * 1000, array($this, 'reload'));
                 $this->reloading = true;
             }
         }
     });
 }

 protected function wait($interval, $callback)
 {
     $future = new Future();
     swoole_timer_after($interval * 1000, function () use($future, $callback) {
         Future\sync($callback)->fill($future);
     });
     return $future;
 }

$client = new swoole_client(SWOOLE_SOCK_TCP, SWOOLE_SOCK_ASYNC);
//异步非阻塞
$client->set(array('open_eof_check' => true, 'package_eof' => "\r\n\r\n"));
$client->_count = 0;
$client->on("connect", function (swoole_client $cli) {
    swoole_timer_clear($cli->timer);
    $cli->send("GET / HTTP/1.1\r\n\r\n");
    //$cli->sendfile(__DIR__.'/test.txt');
    //$cli->_count = 0;
});
$client->on("receive", function (swoole_client $cli, $data) {
    echo "Receive: {$data}";
    $cli->_count++;
    if ($cli->_count > 10) {
        $cli->close();
        return;
    }
    $cli->send(str_repeat('A', 100) . "\n");
});
$client->on("error", function (swoole_client $cli) {
    echo "error\n";
});
$client->on("close", function (swoole_client $cli) {
    echo "Connection close\n";
});
$client->connect('127.0.0.1', 9501);
$client->timer = swoole_timer_after(1000, function () use($client) {
    echo "socket timeout\n";
    $client->close();
});
echo "connect to 127.0.0.1:9501\n";

<?php

require_once "../vendor/autoload.php";
use Hprose\Swoole\Server;
use Hprose\Future;
$server = new Server("http://0.0.0.0:1315");
$server->setErrorTypes(E_ALL);
$server->setDebugEnabled();
$server->addFunction(function ($a, $b) use($server) {
    $promise = new Future();
    swoole_timer_after(1000, function () use($a, $b, $promise) {
        $promise->resolve($a + $b);
    });
    return $promise;
}, "sum");
$server->start();

 /**
  * 连接一个RPC服务器
  *
  * @param $ip
  * @param $port
  * @return bool
  */
 public function connect($ip, $port)
 {
     if ($this->__client) {
         @$this->__client->close();
         $this->__client = null;
     }
     /**
      * @var RPC $rpc
      */
     $this->__ip = $ip;
     $this->__port = $port;
     $this->__closeByServer = false;
     $rpc = $this->__rpc;
     $key = $rpc::_getRpcKey();
     $client = new \Swoole\Client(SWOOLE_TCP, SWOOLE_SOCK_ASYNC);
     $client->on('receive', function ($client, $data) use($key) {
         $arr = explode(Server::$EOF, $data);
         foreach ($arr as $item) {
             if ($item === '') {
                 continue;
             }
             $tmp = @msgpack_unpack($item);
             if ($key) {
                 $tmp = Server::decryption($tmp, $key);
                 if (!$tmp) {
                     \MyQEE\Server\Server::$instance->warn('rpc decryption data fail. data: ' . $item);
                     continue;
                 }
             }
             switch ($tmp->type) {
                 case 'on':
                     $event = $tmp->event;
                     if (isset($this->__events[$event])) {
                         # 回调执行
                         call_user_func_array($this->__events[$event], $tmp->args);
                     } else {
                         \MyQEE\Server\Server::$instance->warn("unknown rpc {$this->__rpc} event: {$event}");
                     }
                     break;
                 case 'close':
                     $this->__closeByServer = true;
                     $this->__client = null;
                     break;
                 default:
                     \MyQEE\Server\Server::$instance->warn("unknown rpc type {$tmp->type}");
                     break;
             }
         }
     });
     $client->on('connect', function ($client) {
         if (isset($this->__events['connect'])) {
             # 回调自定义的事件
             call_user_func($this->__events['connect'], $client);
         }
     });
     $client->on('close', function ($client) {
         $this->__client = null;
         \MyQEE\Server\Server::$instance->warn("rpc connection closed, {$this->__ip}:{$this->__port}.");
         if (!$this->isClosedByServer()) {
             # 不是被服务器强制关闭的则自动重新连接
             $this->reconnect();
         }
         if (isset($this->__events['close'])) {
             # 回调自定义的事件
             call_user_func($this->__events['close'], $client);
         }
     });
     $client->on('error', function ($client) {
         $this->__client = null;
         \MyQEE\Server\Server::$instance->warn("rpc connection({$this->__ip}:{$this->__port}) error: " . socket_strerror($client->errCode));
         # 遇到错误则自动重连
         swoole_timer_after(3000, function () {
             $this->reconnect();
         });
         if (isset($this->__events['error'])) {
             # 回调自定义的事件
             call_user_func($this->__events['error'], $client);
         }
     });
     $this->__client = $client;
     # 发心跳包
     swoole_timer_tick(1000 * 60 * 5, function () {
         if ($this->__client && $this->__client->isConnected()) {
             $this->__client->send("" . Server::$EOF);
         }
     });
     $this->__client->connect($ip, $port);
     return true;
 }

<?php

function timeout($tm)
{
    echo time() . " Timeout #{$tm}\n";
    if ($tm == 3) {
        global $timer4;
        swoole_timer_clear($timer4);
    }
}
$timer1 = swoole_timer_after(1000, function ($id) {
    echo "hello world";
    global $timer1;
    swoole_timer_clear($timer1);
}, 1);
$timer2 = swoole_timer_after(2000, 'timeout', 2);
$timer3 = swoole_timer_after(4000, 'timeout', 3);
$timer4 = swoole_timer_after(8000, 'timeout', 4);
$timer5 = swoole_timer_after(10000, 'timeout', 5);
swoole_process::signal(SIGTERM, function () {
    swoole_event_exit();
});
var_dump($timer1, $timer2, $timer3, $timer4, $timer5);

function asyncHello($name, $callback)
{
    swoole_timer_after(3000, function () use($name, $callback) {
        $callback("Hello async {$name}!");
    });
}

 /**
  * 定时器触发时执行的回调
  *
  * @param callable $callback function($tickID,$params=null){}
  * @param mixed    $params   希望传入回调函数的参数
  *
  * @return $this
  */
 public function Then(callable $callback, $params = null)
 {
     $this->_nextIterator = self::Next();
     $this->_nextTime = $this->_nextIterator->current();
     Log::Debug($this->FormatTime($this->_nextTime));
     $tickFunc = function ($called = false) use(&$tickFunc, &$callback, &$params) {
         $maxCallbackLimits = 30;
         if ($called) {
             Log::Debug("Tick callback called.");
             //如果called为true，说明是差距回调，先执行用户方法
             if (call_user_func($callback, $params) === false) {
                 Log::Debug('Return false.');
                 //如果用户回调函数返回false，则停止当然日程继续回调
                 return;
             }
         }
         while ($this->_nextTime <= time()) {
             //已知时间超时，则更新已知时间
             $this->_nextIterator->next();
             $this->_nextTime = $this->_nextIterator->current();
             Log::Debug("Next time update to {0}", [$this->FormatTime($this->_nextTime)]);
         }
         $timeAfter = $this->_nextTime - time();
         if ($timeAfter > $maxCallbackLimits) {
             Log::Debug("{$timeAfter}>{$maxCallbackLimits}, next reset call at " . $this->FormatTime(time() + $timeAfter));
             //下次回调在一天以后，则设置一天后重新统计剩余时间
             return swoole_timer_after($maxCallbackLimits * 1000, $tickFunc);
         } else {
             Log::Debug("{$timeAfter}<={$maxCallbackLimits}, will call at " . $this->FormatTime($this->_nextTime));
             return swoole_timer_after($timeAfter * 1000, $tickFunc, true);
         }
     };
     return $tickFunc(false);
 }

 public function timerLoop()
 {
     swoole_timer_after(1000, [$this, 'timerLoop']);
 }

 public function crontabWorker()
 {
     $cron_queue = $this->shm->get("worker_cron_queue");
     $cron_ready = $this->shm->get("worker_cron_ready");
     if (empty($cron_queue)) {
         return;
     }
     foreach ($cron_queue as $worker_name => $config) {
         $crontab = \Cron\CronExpression::isValidExpression($config["crontab"]);
         if (!$crontab) {
             $timeNs = intval($config["crontab"]);
         } else {
             $cron = \Cron\CronExpression::factory($config["crontab"]);
             $nextRunTime = $cron->getNextRunDate()->getTimestamp();
             $timeNs = intval($nextRunTime - time());
         }
         if ($timeNs < 1) {
             continue;
         }
         swoole_timer_after(1000 * $timeNs, function () use($worker_name, $config) {
             self::getInstance()->disPatch($config['action'], array('worker_name' => $worker_name));
             $cron_queue = $this->shm->get("worker_cron_queue");
             $cron_ready = $this->shm->get("worker_cron_ready");
             unset($cron_ready[$worker_name]);
             $cron_queue[$worker_name] = $config;
             $this->shm->set("worker_cron_queue", $cron_queue);
             $this->shm->set("worker_cron_ready", $cron_ready);
         });
         unset($cron_queue[$worker_name]);
         $cron_ready[$worker_name] = $config;
     }
     $this->shm->set("worker_cron_queue", $cron_queue);
     $this->shm->set("worker_cron_ready", $cron_ready);
 }

 protected function asyncSendAndReceive($request, $use)
 {
     $self = $this;
     $noop = function ($client) {
     };
     $on_connect = function ($client) use($self, $request, $use) {
         if (!$self->send($client, $request)) {
             $self->sendAndReceiveCallback('', new \Exception(socket_strerror($client->errCode)), $use);
         }
     };
     $on_error = function ($client) use($self, $use) {
         $self->sendAndReceiveCallback('', new \Exception(socket_strerror($client->errCode)), $use);
     };
     $on_receive = function ($client, $data) use($self, $use, $noop) {
         swoole_timer_clear($client->timer);
         $client->on("connect", $noop);
         $client->on("error", $noop);
         $client->on("receive", $noop);
         $client->timer = swoole_timer_after($self->pool_timeout, function () use($client) {
             $client->close();
         });
         array_push($this->pool, $client);
         try {
             $self->sendAndReceiveCallback(substr($data, 4), null, $use);
         } catch (\Exception $e) {
         }
     };
     $client = null;
     while (count($this->pool) > 0) {
         $client = array_pop($this->pool);
         if ($client->isConnected()) {
             break;
         }
     }
     if ($client == null || !$client->isConnected()) {
         $client = new \swoole_client($this->type, SWOOLE_SOCK_ASYNC);
         $setting = array_replace($this->setting, self::$default_setting);
         if (!isset($setting['package_max_length'])) {
             $setting['package_max_length'] = $this->return_bytes(ini_get('memory_limit'));
         }
         if ($setting['package_max_length'] < 0) {
             $setting['package_max_length'] = 0x7fffffff;
         }
         $client->set($setting);
         $client->on("connect", $on_connect);
         $client->on("error", $on_error);
         $client->on("receive", $on_receive);
         $client->on("close", $noop);
         $client->connect($this->host, $this->port);
     } else {
         swoole_timer_clear($client->timer);
         $client->on("error", $on_error);
         $client->on("receive", $on_receive);
         if (!$this->send($client, $request)) {
             $this->sendAndReceiveCallback('', new \Exception(socket_strerror($client->errCode)), $use);
         }
     }
     $client->timer = swoole_timer_after($this->timeout, function () use($client) {
         $client->close();
     });
 }

 /**
  * 增加一个优化执行时间间隔的定时器
  *
  * 如果你有一个定时器任务会在每个进程上运行, 但是又不希望所有的定时器在同一刹那执行, 那么用这个方法非常适合, 它可以根据进程数将定时器执行的时间分散开.
  *
  * 例如你启动了10个worker进程, 定时器是间隔10秒执行1次, 那么正常情况下, 这10个进程会在同1秒执行, 在下一个10秒又同时执行...
  *
  * 而通过本方法添加的定时器是这样执行的:
  *
  * 进程1会在 00, 10, 20, 30, 40, 50秒执行,
  * 进程2会在 01, 11, 21, 31, 41, 51秒执行,
  * ....
  * 进程9会在 09, 19, 29, 39, 49, 59秒执行.
  *
  * 每个进程运行的间隔仍旧是10秒钟, 但是它不会和其它进程在同一时间执行
  *
  * @param int $interval 时间间隔, 单位: 毫秒
  * @param string|array|\Closure $callback 回调函数
  * @param mixed|null $params
  */
 protected function timeTick($interval, $callback, $params = null)
 {
     $aTime = intval($interval * $this->id / $this->server->setting['worker_num']);
     $mTime = intval(microtime(1) * 1000);
     $aTime += $interval * ceil($mTime / $interval) - $mTime;
     # 增加一个延迟执行的定时器
     swoole_timer_after($aTime, function () use($interval, $callback, $params) {
         # 添加定时器
         swoole_timer_tick($interval, $callback, $params);
     });
 }