Реализация пула соединений Swoole MySQL

Статья 84

Речь идет о обучении въезда с Swoole восьмой статьи: достижение пула соединений Swoole MySQL.

Часть 7: Реализация Swoole RPC
Часть 6: Swoole интегрирован в небольшой фреймворк
Часть 5: Многопротокольное многопортовое приложение Swoole
Часть 4: Применение Swoole HTTP
Часть 3: Применение Swoole WebSocket
Часть 2: Применение Swoole Task
Сначала: приложение таймера Swool

Обзор

После получения запросов от читателей, у вас тоже может быть такая ситуация, поэтому я скажу это здесь:

«Брат Лян, в этом году мне 30 лет, и у меня кризис среднего возраста. В последнее время я немного беспокоюсь. Я обнаружил, что, хотя я очень занят на работе, я не чувствую, что мои способности улучшились. У меня есть бесконечные бизнес-коды в течение всего дня. Не так уж хорошо, я знаю, что есть еще много возможностей для улучшения, но у меня не было времени изменить это, главным образом потому, что более поздние проекты находятся под давлением, и они о чтобы снова войти в разработку, что мне делать в этой ситуации?"

Во-первых, я новичок, и если не понравится - не брызгайте, так что выскажу свое мнение:

Приведенное выше описание довольно субъективно, иногда людям свойственно не раскрывать свои способности, а иногда они не сразу проявляют свои способности, но когда достигают определенной стадии, вдруг обнаруживают, вау, они такие сильные .

Конечно, есть много видов способностей, таких как профессиональные способности, способность к быстрому обучению, способность контролировать прогресс и уверенность в себе.

На самом деле, самый быстрый способ улучшить свои способности — это полагаться на рабочую практику.Спокойно спросите себя, присоединились ли вы ко многим группам Даниэля в WeChat, улучшили ли вы свои способности? Чтение и самостоятельное изучение без практики многого не усвоит.

Если вам и приходится давать конкретный план, то это больше делиться внутри команды, потому что перед тем, как поделиться, вы будете полностью готовы, чтобы не выставить себя дураком, когда делитесь, даже если вы иногда знаете, что это не тот случай, когда вы это говорите. .

Ранний обмен может заключаться в чтении рукописи, а более поздний — в обмене без рукописи.

Тогда, скажу еще немного, поставьте себе цель, когда вам будет 30, не учитесь вслепую каждый день, например, цель — технический эксперт, а цель — бизнес-эксперт.

Установите некоторые планы вокруг своих целей.Не думайте, что все остальное придет само собой после ежедневного обучения.Есть также много возможностей и сетевых факторов.

Наконец, если вы действительно чувствуете себя подавленным, измените обстановку и не усложняйте себе задачу.

Начните сегодняшнюю статью, в этой статье реализован пул соединений Swoole MySQL, а код разработан на основе статьи «Реализация Swoole RPC».

Сначала просмотрите содержание предыдущей статьи:

Реализованы HTTP/TCP запросы
Реализованы синхронные/асинхронные запросы
Общий исходный код OnRequest.php, OnReceive.php
Бизнес-логика Order.php возвращает поддельные данные

Основные особенности этой статьи:

Бизнес-логика Order.php возвращает данные в базу данных MySQL.
В задаче включены сопрограммы
Поддерживает конфигурацию базы данных master/slave.
Реализовать пул соединений с базой данных
Внедрить баз данных

код

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]);
    }
}
скопировать код

Задача включает сопрограммы

1. Необходимо добавить две новые конфигурации:


enable_coroutine      = true
task_enable_coroutine = true
скопировать код

2. Изменены параметры обратного вызова


$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; //任务的数据
});
скопировать код

Конфигурация master/slave базы данных

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");
        }
    }
}
скопировать код

База данных 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 = [])
    {
        $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);
скопировать код

расширять

Официальное руководство клиента сопрограммы MySQL:

https://wiki.swoole.com/wiki/page/p-coroutine_mysql.html

Можно попробовать другие способы, предоставленные официалом.

резюме

Демонстрационный код предназначен только для ознакомления, в нем много неточных мест.

Модифицируйте в соответствии с вашими потребностями, такими как проверка бизнес-кода, инкапсуляция метода CURD...

Рекомендовать идеальный продукт, Swoole разработал пул соединения баз данных MySQL (SMPROXY):

https://github.com/louislivi/smproxy

Для приведенной выше демонстрации нужен исходный код, добавьте меня в WeChat. (Меню -> Добавить меня WeChat -> Сканировать меня)