開始今天的文章，這篇文章實作了 Swoole MySQL 連接池，代碼是在《Swoole RPC 的實作》文章的基礎上進行開發的，
先回顧上篇文章的內容：
實作了 HTTP / TCP 請求
實作了 同步 / 異步 請求
分享了 OnRequest.php、OnReceive.php 原始碼
業務邏輯 Order.php 中回傳的是假資料
本篇文章主要的功能點：
業務邏輯 Order.php 中回傳 MySQL 資料庫中的資料，
Task 啟用了協程
支持 主/從 資料庫配置
實作資料庫連接池
實作資料庫 CURD
代碼
Order.php
<?php
if (!defined('SERVER_PATH')) exit("No Access");
class Order
{
public function get_list($uid = 0, $type = 0)
{
//TODO 業務代碼
$rs[0]['order_code'] = '1';
$rs[0]['order_name'] = '訂單1';
$rs[1]['order_code'] = '2';
$rs[1]['order_name'] = '訂單2';
$rs[2]['order_code'] = '3';
$rs[2]['order_name'] = '訂單3';
return $rs;
}
}
修改成：
class Order
{
private $mysql;
private $table;
public function __construct()
{
$pool = MysqlPool::getInstance();
$this->mysql = $pool->get();
$this->table = 'order';
}
public function add($code = '', $name = '')
{
//TODO 驗證
return $this->mysql->insert($this->table, ['code' => $code, 'name' => $name]);
}
public function edit($id = 0, $name='')
{
//TODO 驗證
return $this->mysql->update($this->table, ['name' => $name], ['id' => $id]);
}
public function del($id = 0)
{
//TODO 驗證
return $this->mysql->delete($this->table, ['id' => $id]);
}
public function info($code = '')
{
//TODO 驗證
return $this->mysql->select($this->table, ['code' => $code]);
}
}
Task 啟用協程
一、需要新增兩項配置：
enable_coroutine = true
task_enable_coroutine = true
二、回呼引數發生改變
$serv->on('Task', function ($serv, $task_id, $src_worker_id, $data) {
...
});
修改成：
$serv->on('Task', function ($serv, $task) {
$task->worker_id; //來自哪個`Worker`行程
$task->id; //任務的編號
$task->data; //任務的資料
});
資料庫 主/從 配置
Mysql.php
<?php
if (!defined('SERVER_PATH')) exit("No Access");
$db['default']['pool_size'] = 3; //連接池個數
$db['default']['pool_get_timeout'] = 0.5; //獲取連接池超時時間
$db['default']['timeout'] = 0.5; //資料庫建立連接超時時間
$db['default']['charset'] = 'utf8'; //字符集
$db['default']['strict_type'] = false; //開啟嚴格模式
$db['default']['fetch_mode'] = true; //開啟fetch模式
$config['master'] = $db['default'];
$config['master']['host'] = '127.0.0.1';
$config['master']['port'] = 3306;
$config['master']['user'] = 'root';
$config['master']['password'] = '123456';
$config['master']['database'] = 'demo';
$config['slave'] = $db['default'];
$config['slave']['host'] = '127.0.0.1';
$config['slave']['port'] = 3306;
$config['slave']['user'] = 'root';
$config['slave']['password'] = '123456';
$config['slave']['database'] = 'demo';
資料庫連接池
MysqlPool.php
<?php
if (!defined('SERVER_PATH')) exit("No Access");
class MysqlPool
{
private static $instance;
private $pool;
private $config;
public static function getInstance($config = null)
{
if (empty(self::$instance)) {
if (empty($config)) {
throw new RuntimeException("MySQL config empty");
}
self::$instance = new static($config);
}
return self::$instance;
}
public function __construct($config)
{
if (empty($this->pool)) {
$this->config = $config;
$this->pool = new chan($config['master']['pool_size']);
for ($i = 0; $i < $config['master']['pool_size']; $i++) {
go(function() use ($config) {
$mysql = new MysqlDB();
$res = $mysql->connect($config);
if ($res === false) {
throw new RuntimeException("Failed to connect mysql server");
} else {
$this->pool->push($mysql);
}
});
}
}
}
public function get()
{
if ($this->pool->length() > 0) {
$mysql = $this->pool->pop($this->config['master']['pool_get_timeout']);
if (false === $mysql) {
throw new RuntimeException("Pop mysql timeout");
}
defer(function () use ($mysql) { //釋放
$this->pool->push($mysql);
});
return $mysql;
} else {
throw new RuntimeException("Pool length <= 0");
}
}
}
這里我還準備了一分學習圖和資料，如下：
鏈接：https://pan.baidu.com/s/1v5gm7n0L7TGyejCmQrMh2g 提取碼：x2p5
免費分享，但是X度限制嚴重，如若鏈接失效點擊鏈接或搜索加群 群號518475424，
資料庫 CURD
MysqlDB.php
<?php
if (!defined('SERVER_PATH')) exit("No Access");
class MysqlDB
{
private $master;
private $slave;
private $config;
public function __call($name, $arguments)
{
if ($name != 'query') {
throw new RuntimeException($name."：This command is not supported");
} else {
return $this->_execute($arguments[0]);
}
}
public function connect($config)
{
//主庫
$master = new Swoole\Coroutine\MySQL();
$res = $master->connect($config['master']);
if ($res === false) {
throw new RuntimeException($master->connect_error, $master->errno);
} else {
$this->master = $master;
}
//從庫
$slave = new Swoole\Coroutine\MySQL();
$res = $slave->connect($config['slave']);
if ($res === false) {
throw new RuntimeException($slave->connect_error, $slave->errno);
} else {
$this->slave = $slave;
}
$this->config = $config;
return $res;
}
public function insert($table = '', $data = https://www.cnblogs.com/it-3327/p/[])
{
$fields ='';
$values = '';
$keys = array_keys($data);
foreach ($keys as $k) {
$fields .= "`".addslashes($k)."`, ";
$values .= "'".addslashes($data[$k])."', ";
}
$fields = substr($fields, 0, -2);
$values = substr($values, 0, -2);
$sql = "INSERT INTO `{$table}` ({$fields}) VALUES ({$values})";
return $this->_execute($sql);
}
public function update($table = '', $set = [], $where = [])
{
$arr_set = [];
foreach ($set as $k => $v) {
$arr_set[] = '`'.$k . '` = ' . $this->_escape($v);
}
$set = implode(', ', $arr_set);
$where = $this->_where($where);
$sql = "UPDATE `{$table}` SET {$set} {$where}";
return $this->_execute($sql);
}
public function delete($table = '', $where = [])
{
$where = $this->_where($where);
$sql = "DELETE FROM `{$table}` {$where}";
return $this->_execute($sql);
}
public function select($table = '',$where = [])
{
$where = $this->_where($where);
$sql = "SELECT * FROM `{$table}` {$where}";
return $this->_execute($sql);
}
private function _where($where = [])
{
$str_where = '';
foreach ($where as $k => $v) {
$str_where .= " AND `{$k}` = ".$this->_escape($v);
}
return "WHERE 1 ".$str_where;
}
private function _escape($str)
{
if (is_string($str)) {
$str = "'".$str."'";
} elseif (is_bool($str)) {
$str = ($str === FALSE) ? 0 : 1;
} elseif (is_null($str)) {
$str = 'NULL';
}
return $str;
}
private function _execute($sql)
{
if (strtolower(substr($sql, 0, 6)) == 'select') {
$db = $this->_get_usable_db('slave');
} else {
$db = $this->_get_usable_db('master');
}
$result = $db->query($sql);
if ($result === true) {
return [
'affected_rows' => $db->affected_rows,
'insert_id' => $db->insert_id,
];
}
return $result;
}
private function _get_usable_db($type)
{
if ($type == 'master') {
if (!$this->master->connected) {
$master = new Swoole\Coroutine\MySQL();
$res = $master->connect($this->config['master']);
if ($res === false) {
throw new RuntimeException($master->connect_error, $master->errno);
} else {
$this->master = $master;
}
}
return $this->master;
} elseif ($type == 'slave') {
if (!$this->slave->connected) {
$slave = new Swoole\Coroutine\MySQL();
$res = $slave->connect($this->config['slave']);
if ($res === false) {
throw new RuntimeException($slave->connect_error, $slave->errno);
} else {
$this->slave = $slave;
}
}
return $this->slave;
}
}
}
OnWorkerStart 中呼叫
try {
MysqlPool::getInstance(get_config('mysql'));
} catch (\Exception $e) {
$serv->shutdown();
} catch (\Throwable $throwable) {
$serv->shutdown();
}
客戶端發送請求
<?php
//新增
$demo = [
'type' => 'SW',
'token' => 'Bb1R3YLipbkTp5p0',
'param' => [
'class' => 'Order',
'method' => 'add',
'param' => [
'code' => 'C'.mt_rand(1000,9999),
'name' => '訂單-'.mt_rand(1000,9999),
],
],
];
//編輯
$demo = [
'type' => 'SW',
'token' => 'Bb1R3YLipbkTp5p0',
'param' => [
'class' => 'Order',
'method' => 'edit',
'param' => [
'id' => '4',
'name' => '訂單-'.mt_rand(1000,9999),
],
],
];
//洗掉
$demo = [
'type' => 'SW',
'token' => 'Bb1R3YLipbkTp5p0',
'param' => [
'class' => 'Order',
'method' => 'del',
'param' => [
'id' => '1',
],
],
];
//查詢
$demo = [
'type' => 'SW',
'token' => 'Bb1R3YLipbkTp5p0',
'param' => [
'class' => 'Order',
'method' => 'info',
'param' => [
'code' => 'C4649'
],
],
];
$ch = curl_init();
$options = [
CURLOPT_URL => 'http://10.211.55.4:9509/',
CURLOPT_POST => 1,
CURLOPT_POSTFIELDS => json_encode($demo),
];
curl_setopt_array($ch, $options);
curl_exec($ch);
curl_close($ch);
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/119114.html
標籤：PHP