Skip to content

Commit

Permalink
修改数据库连接池服务
Browse files Browse the repository at this point in the history 
修改数据库连接池服务
  • Loading branch information
@qieangel2013
qieangel2013 committed Aug 11, 2016
1 parent badcc2c commit 14fd2f1
Show file tree
Hide file tree
Showing 3 changed files with 62 additions and 51 deletions.
10 changes: 1 addition & 9 deletions application/library/swoole/socket.php
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,30 +4,22 @@ class swoole_socket extends Model {
public $user;
public function __construct() {
parent::__construct($this->table);
/*if (!isset($this->user)){
$this->user = new swoole_socket();
} */
}
public function getname($userid){
//$where=array('id' =>37936);
$where=array('id' =>$userid);
$result =$this->user->where($where)->select();
}
public static function savefd($fd,$kname='fd'){
//$where=array('id' =>37936);
$redis_con=new phpredis();
$redis_con=new phpredis();
if($redis_con->listSize($kname)){
$redis_con->listPush($kname,$fd,0,1);
}else{
$redis_con->listPush($kname,$fd);
}
//file_put_contents( __DIR__ .'/log.txt' , $fd);
}
public static function getfd($kname='fd'){
$redis_con=new phpredis();
$result=$redis_con->listGet($kname,0,-1);
//$m = file_get_contents( __DIR__ .'/log.txt');
//echo $m;
echo json_encode($result);
}
public static function removefd($fd,$kname='fd'){
Expand Down
2 changes: 1 addition & 1 deletion conf/application.ini
View file
Original file line number Diff line number Diff line change
Expand Up @@ -46,7 +46,7 @@ ftp.config.host="192.168.1.26"
ftp.config.username="root"
ftp.config.password="root"
;redis配置
redis.config.server="127.0.0.1"
redis.config.server="192.168.162.163"
redis.config.port="6379"
;数据库连接池配置
DbServer.async=true
Expand Down
101 changes: 60 additions & 41 deletions server/mysql/DbServer.php
View file
Original file line number Diff line number Diff line change
Expand Up @@ -96,15 +96,13 @@ public function __construct() {
public function onStart($serv)
{
for ($i = 0; $i < $this->pool_size; $i++) {
$db = new mysqli;
$db->connect($this->config['host'],$this->config['user'],$this->config['pwd'],$this->config['name']);
$db = new swoole_mysql;
$db->connect(array('host' => $this->config['host'],'user' => $this->config['user'],'password' => $this->config['pwd'],'database' => $this->config['name']),array(&$this, 'onSQLReady'));
//设置数据库编码
$db->query("SET NAMES '".$this->config['charset']."'");
$db_sock = swoole_get_mysqli_sock($db);
swoole_event_add($db_sock, array(&$this, 'onSQLReady'));
$this->idle_pool[] = array(
'mysqli' => $db,
'db_sock' => $db_sock,
$db->query("SET NAMES '".$this->config['charset']."'",array(&$this, 'doQuery'));
$this->idle_pool[$i] = array(
'db' => $db,
'sock'=>$i,
'fd' => 0,
);
}
Expand All @@ -117,9 +115,21 @@ public function onpipeMessage($serv, $src_worker_id, $data)
//$this->idle_pool=json_decode($data,true);

}
public function onSQLReady($db_sock)
public function onSQLReady($db,$result)
{
$db_res = $this->busy_pool[$db_sock];
if($result){
foreach ($this->idle_pool as $k => $v) {
if($v['db']==$db){
array_unshift($this->wait_queue,$db);
$this->busy_pool[$k]=array(
'db' => $db,
'sock'=>$k,
'fd' => 0,
);
}
}
}
/*$db_res = $this->busy_pool[$db_sock];
$mysqli = $db_res['mysqli'];
$fd = $db_res['fd'];
$data_select=array('status' =>'ok','error'=>0,'errormsg'=>'','result'=>'');
Expand Down Expand Up @@ -151,25 +161,25 @@ public function onSQLReady($db_sock)
$req = array_shift($this->wait_queue);
$this->doQuery($req['fd'], $req['sql']);
}
}
}*/
}

public function onReceive($serv, $fd, $from_id, $data)
{
if($this->isasync){
if (count($this->idle_pool) == 0) {
//if (count($this->idle_pool) == 0) {
//等待队列未满
if (count($this->wait_queue) < $this->wait_queue_max) {
$this->wait_queue[] = array(
'fd' => $fd,
'sql' => $data,
);
} else {
$this->http->send($fd, "request too many, Please try again later.");
}
} else {
$this->doQuery($fd, $data);
}
//if (count($this->wait_queue) < $this->wait_queue_max) {
// $this->wait_queue[] = array(
// 'fd' => $fd,
// 'sql' => $data,
// );
// } else {
// $this->http->send($fd, "request too many, Please try again later.");
//}
// } else {
$this->dosql($fd,$data);
//}
}else{
if($this->multiprocess){
$result = $this->http->task($data);
Expand All @@ -193,6 +203,26 @@ public function onReceive($serv, $fd, $from_id, $data)
}
}

//连接池策略
public function dosql($fd,$data){
if(count($this->wait_queue) > $this->wait_queue_max) {
$this->http->send($fd, "request too many, Please try again later.");
}else{
if (count($this->wait_queue) > 0) {
$db=array_shift($this->wait_queue);
$httpser=$this->http;
$db->query($data,function($link,$result) use($httpser,$fd){
if($result){
$httpser->send($fd,json_encode($result));
array_unshift($this->wait_queue,$db);
}
});

}
}

}

public function onTask($serv, $task_id, $from_id, $sql)
{
if (!self::$link) {
Expand Down Expand Up @@ -225,28 +255,17 @@ public function onTask($serv, $task_id, $from_id, $sql)
return $data;
}

public function doQuery($fd, $sql)
public function doQuery($link,$result)
{
if($result){
//从空闲池中移除
$db = array_pop($this->idle_pool);
/**
* @var mysqli
*/
$mysqli = $db['mysqli'];
for ($i = 0; $i < 2; $i++) {
$result = $mysqli->query($sql, MYSQLI_ASYNC);
if ($result === false) {
if ($mysqli->errno == 2013 or $mysqli->errno == 2006) {
$mysqli->close();
$r = $mysqli->connect();
if ($r === true) continue;
}
foreach ($this->idle_pool as $k => $v) {
if($link==$v['db']){
unset($this->busy_pool[$k]);
}
break;
}
$db['fd'] = $fd;
//加入工作池中
$this->busy_pool[$db['db_sock']] = $db;
return $result;
}
}
public function onFinish($serv, $data)
{
Expand Down

0 comments on commit 14fd2f1

Please sign in to comment.