public function run($worker)
 {
     swoole_timer_tick(1000, function ($timer_id) {
         static $index = 0;
         $index = $index + 1;
         $this->process->push("Hello");
         var_dump($index);
         if ($index == 10) {
             swoole_timer_clear($timer_id);
         }
     });
 }

 public function clean($conn)
 {
     if (isset($conn->timeoutId)) {
         swoole_timer_clear($conn->timeoutId);
         unset($conn->timeoutId);
     }
 }

function timer($id)
{
    echo "Exec: #{$id}\n";
    //如果是After定时器，直接删除即可
    if (isset($afterTimers[$id])) {
        unset($afterTimers[$id]);
    }
    $r = rand(0, 100);
    //删除After定时器，几率：5%
    if ($r > 95) {
        $timerId = array_rand(G::$afterTimers);
        G::log("ClearTick #{$timerId}");
        swoole_timer_clear($timerId);
        unset(G::$afterTimers[$timerId]);
    } elseif ($r > 90) {
        $timerId = array_rand(G::$tickTimers);
        G::log("ClearTick #{$timerId}");
        swoole_timer_clear($timerId);
        unset(G::$tickTimers[$timerId]);
    } elseif ($r > 88) {
        G::newAfterTimer();
    } elseif ($r > 86) {
        G::newTickTimer();
    }
}

 public function send(callable $callback)
 {
     $client = new \swoole_client(SWOOLE_SOCK_UDP, 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) {
         $cli->close();
         $this->calltime = microtime(true) - $this->calltime;
         swoole_timer_clear($this->timer);
         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)) {
         $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'));
             });
         }
     }
 }

function timeout($tm)
{
    echo time() . " Timeout #{$tm}\n";
    if ($tm == 3) {
        global $timer4;
        swoole_timer_clear($timer4);
    }
}

 public function close()
 {
     foreach ($this->pool as $conn) {
         if (isset($conn->timer)) {
             swoole_timer_clear($conn->timer);
             unset($conn->timer);
         }
         if ($conn->isConnected()) {
             $conn->close();
         }
     }
 }

 private function sendUsers()
 {
     $users = $this->getAllUsers();
     foreach ($users as $user) {
         if (isset($this->getUpdateUsers[$user])) {
             $getUpdateUsers = $this->getUpdateUsers[$user];
             unset($this->getUpdateUsers[$user]);
             swoole_timer_clear($getUpdateUsers->timer);
             $getUpdateUsers->completer->complete($users);
         }
     }
 }

 public function run()
 {
     $this->current_num = $this->min_worker_num;
     for ($i = 0; $i < $this->current_num; $i++) {
         $process = new swoole_process(array($this, 'task_run'), false, 2);
         $pid = $process->start();
         $this->process_list[$pid] = $process;
         $this->process_use[$pid] = 0;
     }
     foreach ($this->process_list as $process) {
         swoole_event_add($process->pipe, function ($pipe) use($process) {
             $data = $process->read();
             var_dump($data);
             $this->process_use[$data] = 0;
         });
     }
     swoole_timer_tick(1000, function ($timer_id) {
         static $index = 0;
         $index = $index + 1;
         $flag = true;
         foreach ($this->process_use as $pid => $used) {
             if ($used == 0) {
                 $flag = false;
                 $this->process_use[$pid] = 1;
                 $this->process_list[$pid]->write($index . "Hello");
                 break;
             }
         }
         if ($flag && $this->current_num < $this->max_worker_num) {
             $process = new swoole_process(array($this, 'task_run'), false, 2);
             $pid = $process->start();
             $this->process_list[$pid] = $process;
             $this->process_use[$pid] = 1;
             $this->process_list[$pid]->write($index . "Hello");
             $this->current_num++;
         }
         var_dump($index);
         if ($index == 10) {
             foreach ($this->process_list as $process) {
                 $process->write("exit");
             }
             swoole_timer_clear($timer_id);
             $this->process->exit();
         }
     });
 }

 public function work()
 {
     //是否有队列任务，有的话给worker进程发消息
     foreach ($this->workers as $pid => $worker) {
         \Log::info("队列worker{$pid}启动", [], 'tube.work');
         $data = $worker['tube'];
         if ($data) {
             $worker['process']->write($data);
         }
     }
     swoole_timer_tick(5000, function ($timerId) {
         if (!$this->pheanstalk->getConnection()->isServiceListening()) {
             //现在是一旦队列服务器崩溃的话，处理队列的主进程将退出。当然可以设置成等待，知道队列服务器恢复，只要将下列代码注释
             \Log::emergency("队列服务器崩溃了!TubeTick监听器退出", [], 'tube.tick');
             swoole_timer_clear($timerId);
             return;
         }
     });
 }

 public function run()
 {
     for ($i = 0; $i < $this->worker_num; $i++) {
         $this->process_list[$i] = new swoole_process(array($this, 'task_run'), false, false);
         $this->process_list[$i]->useQueue();
         $this->process_list[$i]->start();
     }
     swoole_timer_tick(1000, function ($timer_id) {
         static $index = 0;
         $index = $index + 1;
         $this->process->push($index . "Hello");
         var_dump($index);
         if ($index == 10) {
             $this->process->push("exit");
             $this->process->push("exit");
             $this->process->push("exit");
             swoole_timer_clear($timer_id);
         }
     });
 }

 public static function loop($timer_id)
 {
     \SysLog::info(__METHOD__, __CLASS__);
     /*
        遍历自己的数组，发现时间超过预定时间段，且该IO的状态依然是未回包状态，则走超时逻辑
     */
     if (empty(self::$event)) {
         \SysLog::info(__METHOD__ . " del event swoole_timer_clear == {$timer_id} ", __CLASS__);
         swoole_timer_clear($timer_id);
     }
     foreach (self::$event as $socket => $e) {
         $now = microtime(true);
         \SysLog::debug(__METHOD__ . " key == {$socket}  now == {$now} timeout == " . $e['timeout'], __CLASS__);
         if ($now > $e['timeout']) {
             self::del($socket);
             $cli = $e['cli'];
             $cli->close();
             call_user_func_array($e['callback'], $e['params']);
         }
     }
 }

 /**
  * 对外提供接口
  *
  * @param  $server      swoole_server对象
  * @param $fd           TCP客户端连接的文件描述符
  * @param $from_id      TCP连接所在的Reactor线程ID
  * @param $data         收到的数据内容
  */
 public function onReceive(swoole_server $server, $fd, $from_id, $data)
 {
     if (empty($data)) {
         // 发送数据给客户端，请求包错误
         $data = array('code' => 500, 'msg' => '非法请求', 'data' => null);
         $server->send($fd, json_encode($data));
     }
     //局域网管理
     $udpClient = $server->connection_info($fd, $from_id);
     if ($udpClient['server_port'] == '9504') {
         echo $data . PHP_EOL;
         switch ($data) {
             case 'stop':
                 echo '服务器关闭: ' . date('Y-m-d H:i:s') . PHP_EOL;
                 $server->shutdown();
                 $server->send($fd, '服务器关闭成功');
                 break;
             case 'reload':
                 echo 'Worker进程重启: ' . date('Y-m-d H:i:s') . PHP_EOL;
                 $server->reload();
                 $server->send($fd, '服务器Worker重启成功');
                 break;
             default:
                 $server->send($fd, '非法请求');
                 break;
         }
     } else {
         $data = json_decode($data, true);
         //任务数据
         $list = $data['list'];
         //请求类型
         $type = $data['type'];
         switch ($type) {
             //添加定时器
             case 'add':
                 //添加定时器
                 $timeInterval = $list['s_interval'];
                 $taskId = $list['s_id'];
                 $timerId = $server->tick($timeInterval, array($this, 'onTimer'), $list);
                 //更新任务
                 Task::updateTimer($taskId, $timerId);
                 $data = array('code' => 200, 'msg' => '添加定时器成功', 'data' => null);
                 $server->send($fd, json_encode($data));
                 break;
                 //修改定时器
             //修改定时器
             case 'edit':
                 //修改url,定时器间隔时间
                 $data = array('code' => 200, 'msg' => '修改定时器成功', 'data' => null);
                 $server->send($fd, json_encode($data));
                 break;
                 //删除定时器
             //删除定时器
             case 'del':
                 echo '删除定时器: ' . date('Y-m-d H:i:s') . PHP_EOL;
                 //任务ID
                 $taskId = $list['s_id'];
                 //定时器ID
                 $timerId = $list['s_timerId'];
                 //删除定时器
                 swoole_timer_clear($timerId);
                 //删除任务
                 Task::delTask($taskId);
                 $data = array('code' => 200, 'msg' => '删除定时器成功', 'data' => null);
                 $server->send($fd, json_encode($data));
                 break;
             default:
                 $data = array('code' => 500, 'msg' => '非法请求', 'data' => null);
                 $server->send($fd, json_encode($data));
                 break;
         }
     }
 }

<?php

swoole_timer_tick(2000, function ($id) {
    echo "tick-1 2000ms\n";
    var_dump($id);
});
usleep(500000);
swoole_timer_tick(2000, function ($id, $params) {
    echo "tick-2 2000ms\n";
    var_dump($id, $params);
    swoole_timer_clear($id);
}, 2);

 protected function asyncSendAndReceive($request, $use)
 {
     $self = $this;
     $client = new \swoole_client($this->type, SWOOLE_SOCK_ASYNC);
     $buffer = "";
     $len = "";
     $client->on("connect", function ($cli) use($self, $request, $use) {
         if (!$self->send($cli, $request)) {
             $self->sendAndReceiveCallback('', new \Exception(socket_strerror($cli->errCode)), $use);
         }
     });
     $client->on("error", function ($cli) use($self, $use) {
         $self->sendAndReceiveCallback('', new \Exception(socket_strerror($cli->errCode)), $use);
     });
     $client->on("receive", function ($cli, $data) use($self, &$buffer, &$len, $use) {
         do {
             if (count($buffer) == 0 || is_string($len)) {
                 $left = 4 - strlen($len);
                 if (strlen($data) < $left) {
                     $len .= $data;
                     return;
                 }
                 $len .= substr($data, 0, $left);
                 $len = unpack("N", $len);
                 $len = $len[1];
                 $n = strlen($data) - $left;
             } else {
                 $left = 0;
                 $n = strlen($data);
             }
             if ($n == 0) {
                 $buffer = "";
                 return;
             }
             if ($len == $n) {
                 $response = $buffer . substr($data, $left);
                 $buffer = "";
                 $len = "";
                 try {
                     $self->sendAndReceiveCallback($response, null, $use);
                 } catch (\Exception $e) {
                 }
                 swoole_timer_clear($cli->timer);
                 $cli->close();
                 return;
             }
             if ($len > $n) {
                 $buffer .= substr($data, $left);
                 $len -= $n;
                 return;
             }
             $response = $buffer . substr($data, $left, $len);
             $buffer = "";
             $data = substr($data, $left + $len);
             $len = "";
             try {
                 $self->sendAndReceiveCallback($response, null, $use);
             } catch (\Exception $e) {
             }
         } while (true);
     });
     $client->on("close", function ($cli) {
     });
     $client->connect($this->host, $this->port);
     $client->timer = swoole_timer_after($this->timeout, function () use($client) {
         $client->close();
     });
 }

 public function clearTimer()
 {
     \swoole_timer_clear($this->_timerId);
 }

<?php

/**
 * swoole提供了类似JavaScript的setInterval/setTimeout异步高精度定时器，粒度为毫秒级。使用也非常简单。
 *
 * - swoole_timer_tick函数就相当于setInterval，是持续触发的
 * - swoole_timer_after函数相当于setTimeout，仅在约定的时间触发一次
 * - swoole_timer_tick/swoole_timer_after函数会返回一个整数，表示定时器的ID
 * - 可以使用 swoole_timer_clear 清除此定时器，参数为定时器ID
 * 
 */
$i = 0;
//每隔2000ms触发一次
swoole_timer_tick(2000, function ($timer_id) {
    global $i;
    if ($i == 4) {
        swoole_timer_clear($timer_id);
        echo "关闭循环计时器\n";
        return;
    }
    ++$i;
    echo "tick-2000ms\n";
});
//3000秒后执行此函数
swoole_timer_after(3000, function () {
    echo "after 3000ms.\n";
});

 public function timerCallback()
 {
     $this->executeCoroutine();
     $this->next();
     swoole_timer_clear($this->_timerId);
 }

 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();
     });
 }

 /**
  * maybe we can clear timers here
  */
 private function __clear_timer()
 {
     $tmid = 0;
     while ($tmid < 5) {
         @swoole_timer_clear($tmid);
         $tmid++;
     }
 }

 public function clearTimeout($timerid)
 {
     return swoole_timer_clear($timerid);
 }