這篇文章僅僅只實現一個 Redis 連接池，篇幅就太少了，順便將前幾篇整合一下。
Demo 中大概包含這些點：
實現 MySQL 連接池
實現 MySQL CURD 方法的定義
實現 Redis 連接池
實現 Redis 方法的定義
滿足 HTTP、TCP、WebSocket 調用
提供 Demo 供測試
調整 目錄結構
HTTP 調用：
實現 讀取 MySQL 中數據的 Demo
實現 讀取 Redis 中數據的 Demo
640?wx_fmt=png
TCP 調用：
實現 讀取 MySQL 中數據的 Demo
實現 讀取 Redis 中數據的 Demo
WebSocket 調用：
實現 每秒展示 API 調用量 Demo
目錄結構
├─ client
│ ├─ http
│ ├── mysql.php //測試 MySQL 連接
│ ├── redis.php //測試 Redis 連接
│ ├─ tcp
│ ├── mysql.php //測試 MySQL 連接
│ ├── redis.php //測試 Redis 連接
│ ├─ websocket
│ ├── index.html //實現 API 調用量展示
├─ controller
│ ├─ Order.php //實現 MySQL CURD
│ ├─ Product.php //實現 Redis 調用
│ ├─ Statistic.php //模擬 API 調用數據
├─ server
│ ├─ config
│ ├── config.php //默認配置
│ ├── mysql.php //MySQL 配置
│ ├── redis.php //Redis 配置
│ ├─ core
│ ├── Common.php //公共方法
│ ├── Core.php //核心文件
│ ├── HandlerException.php //異常處理
│ ├── callback //回調處理
│ ├── OnRequest.php
│ ├── OnReceive.php
│ ├── OnTask.php
│ ├── ...
│ ├── mysql
│ ├── MysqlDB.php
│ ├── MysqlPool.php
│ ├── redis
│ ├── RedisDB.php
│ ├── RedisPool.php
│ ├─ log -- 需要 讀/寫 權限
│ ├── ...
├─ index.php //入口文件
代碼
server/core/redis/RedisPool.php
<?php
if (!defined('SERVER_PATH')) exit("No Access");
class RedisPool
{
private static $instance;
private $pool;
private $config;
public static function getInstance($config = null)
{
if (empty(self::$instance)) {
if (empty($config)) {
throw new RuntimeException("Redis 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) {
$redis = new RedisDB();
$res = $redis->connect($config);
if ($res === false) {
throw new RuntimeException("Failed to connect redis server");
} else {
$this->pool->push($redis);
}
});
}
}
}
public function get()
{
if ($this->pool->length() > 0) {
$redis = $this->pool->pop($this->config['master']['pool_get_timeout']);
if (false === $redis) {
throw new RuntimeException("Pop redis timeout");
}
defer(function () use ($redis) { //釋放
$this->pool->push($redis);
});
return $redis;
} else {
throw new RuntimeException("Pool length <= 0");
}
}
}
這里我還准備了一分學習圖和資料，如下：
鏈接：https://pan.baidu.com/s/1v5gm7n0L7TGyejCmQrMh2g 提取碼：x2p5
免費分享，但是X度限制嚴重，如若鏈接失效點擊鏈接或搜索加群 群號518475424。
server/core/redis/RedisDB.php
<?php
if (!defined('SERVER_PATH')) exit("No Access");
class RedisDB
{
private $master;
private $slave;
private $config;
public function __call($name, $arguments)
{
// TODO 主庫的操作
$command_master = ['set', 'hset', 'sadd'];
if (!in_array($name, $command_master)) {
$db = $this->_get_usable_db('slave');
} else {
$db = $this->_get_usable_db('master');
}
$result = call_user_func_array([$db, $name], $arguments);
return $result;
}
public function connect($config)
{
//主庫
$master = new Swoole\Coroutine\Redis();
$res = $master->connect($config['master']['host'], $config['master']['port']);
if ($res === false) {
throw new RuntimeException($master->errCode, $master->errMsg);
} else {
$this->master = $master;
}
//從庫
$slave = new Swoole\Coroutine\Redis();
$res = $slave->connect($config['slave']['host'], $config['slave']['port']);
if ($res === false) {
throw new RuntimeException($slave->errCode, $slave->errMsg);
} else {
$this->slave = $slave;
}
$this->config = $config;
return $res;
}
private function _get_usable_db($type)
{
if ($type == 'master') {
if (!$this->master->connected) {
$master = new Swoole\Coroutine\Redis();
$res = $master->connect($this->config['master']['host'], $this->config['master']['port']);
if ($res === false) {
throw new RuntimeException($master->errCode, $master->errMsg);
} else {
$this->master = $master;
}
}
return $this->master;
} elseif ($type == 'slave') {
if (!$this->slave->connected) {
$slave = new Swoole\Coroutine\Redis();
$res = $slave->connect($this->config['slave']['host'], $this->config['slave']['port']);
if ($res === false) {
throw new RuntimeException($slave->errCode, $slave->errMsg);
} else {
$this->slave = $slave;
}
}
return $this->slave;
}
}
}
client/http/redis.php
<?php
$demo = [
'type' => 'SW',
'token' => 'Bb1R3YLipbkTp5p0',
'param' => [
'class' => 'Product',
'method' => 'set',
'param' => [
'key' => 'C4649',
'value' => '訂單-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);
client/tpc/redis.php
<?php
class Client
{
private $client;
public function __construct() {
$this->client = new swoole_client(SWOOLE_SOCK_TCP, SWOOLE_SOCK_ASYNC);
$this->client->on('Connect', [$this, 'onConnect']);
$this->client->on('Receive', [$this, 'onReceive']);
$this->client->on('Close', [$this, 'onClose']);
$this->client->on('Error', [$this, 'onError']);
}
public function connect() {
if(!$fp = $this->client->connect("0.0.0.0", 9510, 1)) {
echo "Error: {$fp->errMsg}[{$fp->errCode}]".PHP_EOL;
return;
}
}
public function onConnect() {
fwrite(STDOUT, "測試RPC (Y or N):");
swoole_event_add(STDIN, function() {
$msg = trim(fgets(STDIN));
if ($msg == 'y') {
$this->send();
}
fwrite(STDOUT, "測試RPC (Y or N):");
});
}
public function onReceive($cli, $data) {
echo '[Received]:'.$data;
}
public function send() {
$demo = [
'type' => 'SW',
'token' => 'Bb1R3YLipbkTp5p0',
'param' => [
'class' => 'Product',
'method' => 'get',
'param' => [
'code' => 'C4649'
],
],
];
$this->client->send(json_encode($demo));
}
public function onClose() {
echo "Client close connection".PHP_EOL;
}
public function onError() {
}
}
$client = new Client();
$client->connect();
client/websocket/index.html
<!DOCTYPE html>
<html>
<head>
<meta charset="utf-8">
<meta http-equiv="X-UA-Compatible" content="IE=edge">
<meta name="viewport" content="width=device-width, initial-scale=1">
<meta name="description" content="">
<meta name="keywords" content="">
<title>Demo</title>
<script src="https://cdn.bootcss.com/jquery/3.4.1/jquery.js"></script>
<script src="http://echarts.baidu.com/gallery/vendors/echarts/echarts.min.js"></script>
</head>
<body>
<!-- 為ECharts准備一個具備大小（寬高）的Dom -->
<div id="main" style="width: 900px;height:400px;"></div>
<script type="text/javascript">
if ("WebSocket" in window) {
// 基於准備好的dom，初始化echarts實例
var myChart = echarts.init(document.getElementById('main'));
var wsServer = 'ws://10.211.55.4:9509';
var ws = new WebSocket(wsServer);
ws.onopen = function (evt) {
if (ws.readyState == 1) {
console.log('WebSocket 連接成功...');
} else {
console.log('WebSocket 連接失敗...');
}
if (ws.readyState == 1) {
ws.send('開始請求...');
} else {
alert('WebSocket 連接失敗');
}
};
ws.onmessage = function (evt) {
console.log('Retrieved data from server: ' + evt.data);
var evt_data = jQuery.parseJSON(evt.data);
myChart.setOption({
xAxis: {
data : evt_data.time
},
series: [{
data: evt_data.value
}]
});
};
ws.onerror = function (evt) {
alert('WebSocket 發生錯誤');
console.log(evt);
};
ws.onclose = function() {
alert('WebSocket 連接關閉');
console.log('WebSocket 連接關閉...');
};
// 指定圖表的配置項和數據
$.ajax({
url : 'http://10.211.55.4:9509/', // 請求url
type : "post", // 提交方式
dataType : "json", // 數據類型
data : {
'type' : 'SW',
'token' : 'Bb1R3YLipbkTp5p0',
'param' : {
'class' : 'Statistic',
'method' : 'init'
}
},
beforeSend:function() {
},
success : function(rs) {
if (rs.code != 1) {
alert('獲取數據失敗');
} else {
var option = {
title: {
text: 'API 調用量',
x:'center'
},
tooltip: {
trigger: 'axis',
axisPointer: {
animation: false
}
},
xAxis: {
type : 'category',
data : rs.data.time
},
yAxis: {
type: 'value',
boundaryGap: [0, '100%'],
name: '使用量',
splitLine: {
show: false
}
},
series: [{
name: '使用量',
type: 'line',
showSymbol: false,
hoverAnimation: false,
data: rs.data.value
}]
};
// 使用剛指定的配置項和數據顯示圖表。
if (option && typeof option === "object") {
myChart.setOption(option, true);
}
}
},
error : function(){
alert('服務器請求異常');
}
});
} else {
alert("您的瀏覽器不支持 WebSocket!");
}
</script>
</body>
</html>
還涉及到，OnMessage.php、OnTask.php 、OnWorkerStart.php 等，就不貼代碼了。