Swoole Redis 鏈接池的實現

時間 2020-05-26

標籤 swoole redis 鏈接實現欄目 Redis 简体版

原文原文鏈接

概述

這是關於 Swoole 入門學習的第九篇文章：Swoole Redis 鏈接池的實現。javascript

收到讀者反饋，「亮哥，文章能多點圖片嗎？就是將運行結果以圖片的形式展現...」php

我我的以爲這是比較懶、動手能力差的表現，恩... 要勤快些。css

但誰讓文章是寫給大家看的那，我之後儘可能文章寫的圖文並茂一點。html

上篇文章分享了 MySQL 鏈接池，這篇文章我們來分享下 Redis 鏈接池。java

在上篇文章的基礎上進行簡單調整便可，將實例化 MySQL 的地方，修改爲實例化 Redis 便可，還要注意一些方法的調整。mysql

這篇文章僅僅只實現一個 Redis 鏈接池，篇幅就太少了，順便將前幾篇整合一下。jquery

大概 Demo 中包含這些點：web

實現 MySQL 鏈接池
實現 MySQL CURD 方法的定義
實現 Redis 鏈接池
實現 Redis 方法的定義
知足 HTTP、TCP、WebSocket 調用
提供 Demo 供測試
調整目錄結構

HTTP 調用：ajax

實現讀取 MySQL 中數據的 Demo
實現讀取 Redis 中數據的 Demo

TCP 調用：redis

實現讀取 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");
        }
    }
}

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 等，就不貼代碼了。