swoole异步mysql demo

在 2015年11月13日2015年11月16日上张贴由 BestLove

<?php 
class asyncmysql
{
    protected $poolSize = 20;
    protected $freePool = array();
    protected $busyPool = array();
    protected $waitQueue = array();
    protected $waitSize = 50;
    protected $serv;

    public function run()
    {
        $serv = new swoole_server("127.0.0.1", 9501);
        $serv->on('connect', array($this, 'onConnect'));
        $serv->on('receive', array($this, 'onReceive'));
        $serv->on('close', array($this, 'onClose'));
        $serv->on('workerstart', array($this, 'onWorkerStart'));
        $serv->start();
    }

    public function onConnect($serv, $fd)
    {
    }

    public function onClose($serv, $fd)
    {
    }

    public function onWorkerStart($serv, $work_id)
    {
        $this->serv = $serv;
        for ($i=0; $i < $this->poolSize; $i++) {
            $db = new mysqli();
            $db->connect('localhost', 'root', '123456', 'swooletest');
            //swoole获取mysqli的socket文件描述符，需要php有mysqlnd、mysqli扩展
            //swoole编译时需要加入 --enable-async-mysql
            $db_sock = swoole_get_mysqli_sock($db);
            //将mysqli的socket加入到事件监听
            swoole_event_add($db_sock, array($this, 'query'));
            $this->freePool[]=array(
                'db' => $db,
                'db_sock' => $db_sock,
                'fd' => 0
            );
        }
    }

    public function onReceive($serv, $fd, $from_id, $sql)
    {
        if (!empty($this->freePool)) {
            $this->preQuery($fd, $sql);
        } else {
            if (count($this->waitQueue) >= $this->waitSize) {
                $serv->send($fd, '服务器繁忙，请稍后再试！');
            } else {
                $this->waitQueue[]=array(
                    'fd' => $fd,
                    'sql' => $sql
                );
            }
        }
    }

    public function preQuery($fd, $sql)
    {
        $dbArray = array_pop($this->freePool);
        $db = $dbArray['db'];
        for ($i=0; $i < 2; $i++) { 
            //mysqli异步query，并不是真正的执行
            $rs = $db->query($sql, MYSQLI_ASYNC);
            if ($rs === false) {
                //尝试一次超时重连
                if ($db->errno == 2013 || $db->errno == 2006) {
                    $db->close();
                    $r = $db->connect();
                    if ($r === true) continue;
                }
            }
            break;
        }
        $dbArray['fd'] = $fd;
        $this->busyPool[$dbArray['db_sock']] = $dbArray;
    }

    public function query($db_sock)
    {
        $dbArray = $this->busyPool[$db_sock];
        $db = $dbArray['db'];
        $fd = $dbArray['fd'];
        //执行mysqli异步查询
        $rs = $db->reap_async_query();
        if ($rs === true) {
            $this->serv->send($fd, '操作成功！'); 
        } elseif ($rs === false) {
            $this->serv->send($fd, '操作失败！');
        } else {
            $rs = $rs->fetch_all(MYSQLI_ASSOC);
            $this->serv->send($fd, json_encode($rs, JSON_UNESCAPED_UNICODE));
            unset($rs);
        }
        $this->freePool[]=$dbArray;
        unset($this->busyPool[$db_sock]);
        //如果等待队列中有未执行的SQL，循环所有的空闲连接执行SQL
        if (count($this->waitQueue) > 0) {
            $freeCount = count($this->freePool);
            for ($i=0; $i < $freePool; $i++) { 
                $waitArray = array_shift($this->waitQueue);
                $this->preQuery($waitArray['fd'], $waitArray['sql']);
            }
        }
    }
}

$test = new asyncmysql();
$test->run();

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

<?php

class asyncmysql

{

protected $poolSize = 20;

protected $freePool = array();

protected $busyPool = array();

protected $waitQueue = array();

protected $waitSize = 50;

protected $serv;

public function run()

{

$serv = new swoole_server("127.0.0.1", 9501);

$serv->on('connect', array($this, 'onConnect'));

$serv->on('receive', array($this, 'onReceive'));

$serv->on('close', array($this, 'onClose'));

$serv->on('workerstart', array($this, 'onWorkerStart'));

$serv->start();

}

public function onConnect($serv, $fd)

{

}

public function onClose($serv, $fd)

{

}

public function onWorkerStart($serv, $work_id)

{

$this->serv = $serv;

for ($i=0; $i < $this->poolSize; $i++) {

$db = new mysqli();

$db->connect('localhost', 'root', '123456', 'swooletest');

//swoole获取mysqli的socket文件描述符，需要php有mysqlnd、mysqli扩展

//swoole编译时需要加入 --enable-async-mysql

$db_sock = swoole_get_mysqli_sock($db);

//将mysqli的socket加入到事件监听

swoole_event_add($db_sock, array($this, 'query'));

$this->freePool[]=array(

'db' => $db,

'db_sock' => $db_sock,

'fd' => 0

);

}

public function onReceive($serv, $fd, $from_id, $sql)

{

if (!empty($this->freePool)) {

$this->preQuery($fd, $sql);

} else {

if (count($this->waitQueue) >= $this->waitSize) {

$serv->send($fd, '服务器繁忙，请稍后再试！');

} else {

$this->waitQueue[]=array(

'fd' => $fd,

'sql' => $sql

);

}

public function preQuery($fd, $sql)

{

$dbArray = array_pop($this->freePool);

$db = $dbArray['db'];

for ($i=0; $i < 2; $i++) {

//mysqli异步query，并不是真正的执行

$rs = $db->query($sql, MYSQLI_ASYNC);

if ($rs === false) {

//尝试一次超时重连

if ($db->errno == 2013 || $db->errno == 2006) {

$db->close();

$r = $db->connect();

if ($r === true) continue;

}

break;

}

$dbArray['fd'] = $fd;

$this->busyPool[$dbArray['db_sock']] = $dbArray;

}

public function query($db_sock)

{

$dbArray = $this->busyPool[$db_sock];

$db = $dbArray['db'];

$fd = $dbArray['fd'];

//执行mysqli异步查询

$rs = $db->reap_async_query();

if ($rs === true) {

$this->serv->send($fd, '操作成功！');

} elseif ($rs === false) {

$this->serv->send($fd, '操作失败！');

} else {

$rs = $rs->fetch_all(MYSQLI_ASSOC);

$this->serv->send($fd, json_encode($rs, JSON_UNESCAPED_UNICODE));

unset($rs);

}

$this->freePool[]=$dbArray;

unset($this->busyPool[$db_sock]);

//如果等待队列中有未执行的SQL，循环所有的空闲连接执行SQL

if (count($this->waitQueue) > 0) {

$freeCount = count($this->freePool);

for ($i=0; $i < $freePool; $i++) {

$waitArray = array_shift($this->waitQueue);

$this->preQuery($waitArray['fd'], $waitArray['sql']);

}

$test = new asyncmysql();

$test->run();

swoole websocket demo

在 2015年11月10日2015年11月10日上张贴由 BestLove

<?php 
//创建websocket服务器对象，监听0.0.0.0:9502端口
$ws = new swoole_websocket_server("0.0.0.0", 9502);

//监听WebSocket连接打开事件
$ws->on('open', function ($ws, $request) {
    // var_dump($request->fd, $request->get, $request->server);
    foreach($ws->connections as $fd) {
        $ws->push($fd, "你好，有人加入聊天。");
    }
    // $ws->push($request->fd, "你好，有人加入聊天。\n");
});

//监听WebSocket消息事件
$ws->on('message', function ($ws, $frame) {
    // echo "Message: {$frame->data}\n";
    foreach($ws->connections as $fd) {
        $ws->push($fd, "有人说: {$frame->data}");
    }
    // $ws->push($frame->fd, "server: {$frame->data}");
});

//监听WebSocket连接关闭事件
$ws->on('close', function ($ws, $fd) {
    echo "client-{$fd} is closed\n";
});

$ws->start();
?>

<?php

//创建websocket服务器对象，监听0.0.0.0:9502端口

$ws = new swoole_websocket_server("0.0.0.0", 9502);

//监听WebSocket连接打开事件

$ws->on('open', function ($ws, $request) {

// var_dump($request->fd, $request->get, $request->server);

foreach($ws->connections as $fd) {

$ws->push($fd, "你好，有人加入聊天。");

}

// $ws->push($request->fd, "你好，有人加入聊天。\n");

});

//监听WebSocket消息事件

$ws->on('message', function ($ws, $frame) {

// echo "Message: {$frame->data}\n";

foreach($ws->connections as $fd) {

$ws->push($fd, "有人说: {$frame->data}");

}

// $ws->push($frame->fd, "server: {$frame->data}");

});

//监听WebSocket连接关闭事件

$ws->on('close', function ($ws, $fd) {

echo "client-{$fd} is closed\n";

});

$ws->start();

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>websocket</title>
</head>
<body>

测试一下啊：<input type="text" value="" id="test">&nbsp;
            <input type="button" value="发送" id="bt">
<div id="div1"></div>
<script>
    var wsServer = 'ws://192.168.22.228:9502';
    var websocket = new WebSocket(wsServer);
    websocket.onopen = function (evt) {
        console.log("Connected to WebSocket server.");
    };

    websocket.onclose = function (evt) {
        console.log("Disconnected");
    };

    websocket.onmessage = function (evt) {
        // console.log('Retrieved data from server: ' + evt.data);
        div1.innerHTML+=evt.data+'<br>';
    };

    websocket.onerror = function (evt, e) {
        console.log('Error occured: ' + evt.data);
    };
    
    var bt = document.getElementById("bt");
    var test = document.getElementById("test");
    bt.onclick=function(){
        websocket.send(test.value);
    };
</script>
</body>
</html>

<!DOCTYPE html>

<head>

<title>websocket</title>

</head>

<body>

测试一下啊：<input type="text" value="" id="test"> 

var wsServer = 'ws://192.168.22.228:9502';

var websocket = new WebSocket(wsServer);

websocket.onopen = function (evt) {

console.log("Connected to WebSocket server.");

};

websocket.onclose = function (evt) {

console.log("Disconnected");

};

websocket.onmessage = function (evt) {

// console.log('Retrieved data from server: ' + evt.data);

div1.innerHTML+=evt.data+'<br>';

};

websocket.onerror = function (evt, e) {

console.log('Error occured: ' + evt.data);

};

var bt = document.getElementById("bt");

var test = document.getElementById("test");

bt.onclick=function(){

websocket.send(test.value);

};

</script>

</body>

</html>

<?php
//创建websocket服务器对象，监听0.0.0.0:9502端口
$ws = new swoole_websocket_server("0.0.0.0", 9502);
//设置异步任务的工作进程数量
$ws->set(array('task_worker_num' => 4));

//监听WebSocket连接打开事件
$ws->on('open', function ($ws, $request) {
    $ws->task("你好，有人加入聊天。");
});

//监听WebSocket消息事件
$ws->on('message', function ($ws, $frame) {
    $ws->task("有人说: {$frame->data}");
});

//监听WebSocket连接关闭事件
$ws->on('close', function ($ws, $fd) {
    echo "client-{$fd} is closed\n";
});

//处理异步任务
$ws->on('task', function ($ws, $task_id, $from_id, $data) {
    foreach ($ws->connections as $fd) {
        $ws->push($fd, $data);
    }
    //finish操作是可选的
    $ws->finish("ok");
});

//处理异步任务的结果
$ws->on('finish', function ($ws, $task_id, $data) {
    echo $data . PHP_EOL;
});

$ws->start();

<?php

//创建websocket服务器对象，监听0.0.0.0:9502端口

$ws = new swoole_websocket_server("0.0.0.0", 9502);

//设置异步任务的工作进程数量

$ws->set(array('task_worker_num' => 4));

//监听WebSocket连接打开事件

$ws->on('open', function ($ws, $request) {

$ws->task("你好，有人加入聊天。");

});

//监听WebSocket消息事件

$ws->on('message', function ($ws, $frame) {

$ws->task("有人说: {$frame->data}");

});

//监听WebSocket连接关闭事件

$ws->on('close', function ($ws, $fd) {

echo "client-{$fd} is closed\n";

});

//处理异步任务

$ws->on('task', function ($ws, $task_id, $from_id, $data) {

foreach ($ws->connections as $fd) {

$ws->push($fd, $data);

}

//finish操作是可选的

$ws->finish("ok");

});

//处理异步任务的结果

$ws->on('finish', function ($ws, $task_id, $data) {

echo $data . PHP_EOL;

});

$ws->start();

swoole数据库连接复用demo

在 2015年11月9日2015年11月9日上张贴由 BestLove

<?php
require __DIR__ . '/db.php';
$dbPool = array();
$poolSize = 20;
for ($i=0; $i < $poolSize; $i++) { 
    $dbPool[$i]= new Db();
}

$serv = new swoole_server("127.0.0.1", 9501);
// $serv->set(array(
//     'worker_num' => 8,   //工作进程数量
//     'daemonize' => true, //是否作为守护进程
// ));
$serv->on('connect', function ($serv, $fd){
    echo "Client:Connect.\n";
});
$serv->on('receive', function ($serv, $fd, $from_id, $sql) {
    global $dbPool;
    $rs = false;
    if (!empty($dbPool)) {
        $db = array_pop($dbPool);
        $rs = $db->execute($sql);
        array_push($dbPool, $db);
    }
    if ($rs) {
        $serv->send($fd, 'ok');
    } else {
        $serv->send($fd, 'not ok');
    }
    $serv->close($fd);
});
$serv->on('close', function ($serv, $fd) {
    echo "Client: Close.\n";
});
$serv->start();

<?php

require __DIR__ . '/db.php';

$dbPool = array();

$poolSize = 20;

for ($i=0; $i < $poolSize; $i++) {

$dbPool[$i]= new Db();

}

$serv = new swoole_server("127.0.0.1", 9501);

// $serv->set(array(

// 'worker_num' => 8, //工作进程数量

// 'daemonize' => true, //是否作为守护进程

// ));

$serv->on('connect', function ($serv, $fd){

echo "Client:Connect.\n";

});

$serv->on('receive', function ($serv, $fd, $from_id, $sql) {

global $dbPool;

$rs = false;

if (!empty($dbPool)) {

$db = array_pop($dbPool);

$rs = $db->execute($sql);

array_push($dbPool, $db);

}

if ($rs) {

$serv->send($fd, 'ok');

} else {

$serv->send($fd, 'not ok');

}

$serv->close($fd);

});

$serv->on('close', function ($serv, $fd) {

echo "Client: Close.\n";

});

$serv->start();

db类下载地址：

http://www.fengfengphp.com/yaf/db.zip

唐晓锋–BLOG

This is just a personal note.

月份：2015年11月

swoole异步mysql demo

swoole websocket demo

swoole数据库连接复用demo