参考于Git地址:https://github.com/LinkedDestiny/swoole-doc
第三章:Timer定时器、心跳检测及Task进阶实例:mysql连接池
Swoole提供了一个内置的Timer定时器功能,通过函数[addtimer]即可在Swoole中添加一个定时器,该定时器会在建立之后,按照预先设定好的时间间隔,每到对应的时间就会调用一次回调函数[onTimer]。
简单示例如下:
```php
$this->serv->on('Timer', array($this, 'onTimer'));
public function onWorkerStart( $serv , $worker_id) {
// 在Worker进程开启时绑定定时器
// 只有当worker_id为0时才添加定时器,避免重复添加
if( $worker_id == 0 ) {
$serv->addtimer(500);
$serv->addtimer(1000);
$serv->addtimer(1500);
}
}
public function onTimer($serv, $interval) {
switch( $interval ) {
case 500: { //
echo "Do Thing A at interval 500\n";
break;
}
case 1000:{
echo "Do Thing B at interval 1000\n";
break;
}
case 1500:{
echo "Do Thing C at interval 1500\n";
break;
}
}
}
```
可以看到,在[onWorkerStart]回调函数中,通过addtimer添加了三个定时器,时间间隔分别为500、1000、1500。而在onTimer回调中,正好通过间隔的不同来区分不同的定时器回调,从而执行不同的操作。
需要注意的是,在上述示例中,当1000ms的定时器被触发时,500ms的定时器同样会被触发,但是不能保证会在1000ms定时器前触发还是后触发,因此需要注意,定时器中的操作不能依赖其他定时器的执行结果。
在慕课的聊天室案例中建立websocket_server服务器实现定时器
class Ws {
private $HOST = "0.0.0.0";
private $PORT = 8812;
public $ws = null;
public function __construct(){
$this->ws = new swoole_websocket_server($this->HOST, $this->PORT);
$this->ws->set(array(
'worker_num' =>2,
'task_worker_num' =>2,
));
$this->ws->on("open", [$this, 'onOpen']);
$this->ws->on("message", [$this, 'onMessage']);
$this->ws->on("task", [$this, 'onTask']);
$this->ws->on("finish", [$this, 'onFinish']);
$this->ws->on("close", [$this, 'onClose']);
$this->ws->start();
}
/**
*监听ws连接事件
*@param $ws
*@param $request
*/
public function onOpen($ws, $request){
var_dump($request->fd);
if($request->fd == 1) {
//每2秒执行,定时触发
swoole_timer_tick(2000, function($timer_id){
echo "2s:timeId:{$timer_id}";
});
}
}
/**
*监听ws消息事件(收到客户端消息)
*@param $ws
*@param $frame
*/
public function onMessage($ws, $frame){
echo "ser-push-message:{$frame->data}\n";
// todo 10s
$data = [
'task' => 1,
'fd' =>$frame->fd,
];
//$ws->task($data);
//在指定的时间后执行函数
swoole_timer_after(5000, function() use($ws, $frame){
echo "5s-after\n";
$ws->push($frame->fd, "server-time-after:");
});
$ws->push($frame->fd, "server-push:".date("Y-m-d H:i:s"));
}
/**
*onTask(在task_worker进程内被调用)
*@param $serv
*@param $taskId
*@param $workerId
*@param $data
*/
public function onTask($serv, $taskId, $workerId, $data){
print_r($data);
//耗时场景 10s
sleep(3);
return "on task finish"; //告诉worker
}
/**
*onFinish
*@param $serv
*@param $taskId
*@param $data
*/
public function onFinish($serv, $taskId, $data) {
//此处的$data 为onTask中return的内容
echo "taskId:{$taskId}\n";
echo "finish-data-success:{$data}";
}
/**
*close
*@param $ws
*@param $fd
*/
public function onClose($ws, $fd){
echo "clientid:{$fd}\n";
}
}
$ojb = new Ws();
心跳检测
可以使用Timer定时器功能可以实现发送心跳包的功能。事实上,Swoole已经内置了心跳检测功能,能自动close掉长时间没有数据来往的连接。
而开启心跳检测功能,只需要设置
[heartbeat_check_interval]
[heartbeat_idle_time]
```php
$this->serv->set(
array(
'heartbeat_check_interval' => 60,
'heartbeat_idle_time' => 600,
)
);
```
其中heartbeat_idle_time的默认值是heartbeat_check_interval的两倍。
在设置这两个选项后,swoole会在内部启动一个线程,每隔heartbeat_check_interval秒后遍历一次全部连接,检查最近一次发送数据的时间和当前时间的差,如果这个差值大于heartbeat_idle_time,则会强制关闭这个连接,并通过回调onClose通知Server进程。
结合之前的Timer功能,如果我们想维持连接,就设置一个略小于如果这个差值大于heartbeat_idle_time的定时器,在定时器内向所有连接发送一个心跳包。如果收到心跳回应,则判断连接正常,如果没有收到,则关闭这个连接或者再次尝试发送。
Task进阶:MySQL连接池
在PHP中,访问MySQL数据库往往是性能提升的瓶颈。而MySQL连接池我想大家都不陌生,这是一个很好的提升数据库访问性能的方式。传统的MySQL连接池,是预先申请一定数量的连接,每一个新的请求都会占用其中一个连接,请求结束后再将连接放回池中,如果所有连接都被占用,新来的连接则会进入等待状态。
使用Swoole实现一个连接池:
首先,Swoole允许开启一定量的Task Worker进程,我们可以让每个进程都拥有一个MySQL连接,并保持这个连接,这样,我们就创建了一个连接池。
其次,设置swoole的[dispatch_mode]为抢占模式(主进程会根据Worker的忙闲状态选择投递,只会投递给处于闲置状态的Worker)。这样,每个task都会被投递给闲置的Task Worker。这样,我们保证了每个新的task都会被闲置的Task Worker处理,如果全部Task Worker都被占用,则会进入等待队列。
```php
public function onWorkerStart( $serv , $worker_id) {
echo "onWorkerStart\n";
// 判定是否为Task Worker进程
if( $worker_id >= $serv->setting['worker_num'] ) {
$this->pdo = new PDO(
"mysql:host=localhost;port=3306;dbname=Test",
"root",
"123456",
array(
PDO::MYSQL_ATTR_INIT_COMMAND => "SET NAMES 'UTF8';",
PDO::ATTR_ERRMODE => PDO::ERRMODE_EXCEPTION,
PDO::ATTR_PERSISTENT => true
)
);
}
}
```
首先,在每个Task Worker进程中,创建一个MySQL连接。这里我选用了PDO扩展。
```php
public function onReceive( swoole_server $serv, $fd, $from_id, $data ) {
$sql = array(
'sql'=>'select * from Test where pid > ?',
'param' => array(
0
),
'fd' => $fd
);
$serv->task( json_encode($sql) );
}
```
其次,在需要的时候,通过[task]()函数投递一个任务(也就是发起一次SQL请求)
```php
public function onTask($serv,$task_id,$from_id, $data) {
$sql = json_decode( $data , true );
$statement = $this->pdo->prepare($sql['sql']);
$statement->execute($sql['param']);
$result = $statement->fetchAll(PDO::FETCH_ASSOC);
$serv->send( $sql['fd'],json_encode($result));
return true;
}
```
最后,在onTask回调中,根据请求过来的SQL语句以及相应的参数,发起一次MySQL请求,并将获取到的结果通过send发送给客户端(或者通过return返回给Worker进程)。而且,这样的一次MySQL请求还不会阻塞Worker进程,Worker进程可以继续处理其他的逻辑。