swoole4 создает высокопроизводительный пул соединений с базой данных Coroutine.

PHP Swoole

Хранилище кодового облака кода:git ee.com/Тан Цзяцзюнь/М…

Репозиторий кода:GitHub.com/as BEC TJ/ Это я…

предисловие

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

Во-первых, основная концепция пула соединений с базой данных.

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

  • Мультиплексирование соединений, различные соединения запросов могут быть помещены обратно в пул, ожидая выделения и вызова следующего запроса.
  • Количество подключений обычно поддерживается между максимальным и минимальным значениями min-max.
  • Повторное использование неиспользуемых соединений
  • Он может противостоять определенной степени высокого параллелизма, то есть, когда все соединения в пуле запрашиваются одновременно одновременно, запрос, который не может получить соединение, может ждать освобождения других соединений.

После суммирования нескольких функций базовый пул соединений должен примерно реализовывать следующие функции:

 

  1. Создать соединение: после запуска пула соединений инициализируется определенное бездействующее соединение, которое указывается как минимальное минимальное соединение. Когда пул соединений пуст и его недостаточно, создайте новое соединение и поместите его в пул, но оно не может превышать указанное максимальное количество соединений max.
  2. Освобождение соединения: после использования каждого соединения необходимо вызвать метод освобождения, чтобы поместить соединение обратно в пул для использования другими программами или запросами.
  3. Распределение соединений: пул соединений использует методы pop и push для одноранговой постановки в очередь и выделения из очереди, а также повторного использования. Он может реализовать блокировку распределения, то есть, когда пул пуст и количество созданных больше max, блокировать на определенный период времени и ждать освобождения соединения других запросов, и возвращать null, если тайм-аут превышен.
  4. Управление соединениями: для соединений в пуле соединений регулярно проверяйте активные и освобождайте незанятые соединения и т. д.

Во-вторых, реализация длинного соединения с базой данных Fpm+.

  1. Использование fpm для достижения: Например, вы хотите создать пул из 100 соединений, открыть 100 простаивающих fpm, а затем каждое соединение fpm является длинным соединением с базой данных. Как правило, элемент конфигурации pm.max_spare_servers = 8 предназначен для поддержания незанятого количества пулов соединений, а затем pm.max_children = 50 — это максимальное количество соединений. Количество процессов такое же, как и у fpm.

3. Реализация на основе swoole

  • Краткое введение в swoole (более подробную информацию см. на официальном сайте swoole)

swoole — это движок или расширение для PHP для реализации асинхронного сетевого взаимодействия, которое реализует многие вещи, которых нет в традиционном PHP-fpm, например, асинхронный клиент, асинхронный ввод-вывод, резидентную память, сопрограмму и т. д., одно за другим отличное расширение, среди какие концепции, такие как асинхронность и сопрограммы, могут применяться к сценариям с высокой степенью параллелизма. Недостаток в том, что порог для документации и входа относительно высок, и необходимо устраивать ямы. Прикрепите запущенный процесс и схему структуры процесса swoole:

Запустите блок-схему

Диаграмма архитектуры процесса/потока

  • Соображения, основанные на swoole реальности

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

1. Программа использует коммуникационный канал channel сопрограммы (похожий на go's chan), в котором swoole2 не поддерживает ожидание таймаута в chan->pop($timeout), поэтому необходимо использовать версию swoole4

2. При использовании расширения сопрограммы swoole нельзя устанавливать такие расширения, как xdebug, иначе будет сообщено об ошибке. Официальное описание:wiki.Say Me Oh.com/wiki/Afraid/6…, и обратитесь к следующему, чтобы узнать больше об использовании и внимании к сопрограммам swoole:wiki.Say Me Oh.com/wiki/Afraid/7…

3. Среда, используемая автором: PHP 7.1.18 и swoole4 в качестве среды для данной разработки.

  • Метод, основанный на реалистичном пуле соединений swoole

Прежде всего, на этот раз swoole используется для реализации пула соединений, и применяются следующие технологии или концепции swoole.

1. Подключите пул переменных, здесь вы можете увидеть массив или очередь, используя характеристики резидентной памяти глобальной переменной Swoole, пока переменная не активна Unset drop, объект соединения в массиве или очереди может быть сохранен, не издан. Основная ссылка:вики Я сказал о .com/вики/страх/страх...

2. Сопрограммы. Сопрограммы — это чисто пользовательские потоки, которые переключаются кооперативным, а не упреждающим образом. Прежде всего, этот пул соединений использует сопрограммы в двух местах:

  • Одним из них является клиент сопрограммы mysql, зачем использовать клиент сопрограммы, потому что, если вы используете PDO синхронного клиента, даже если в одном процессе есть сотни пулов соединений, рабочий процесс swoole использует обычный метод PDO.Независимо от того, сколько одновременных запросов каждый запрос может ожидать только завершения выполнения предыдущего запроса, прежде чем рабочий процесс обработает следующий запрос, даже если он здесь заблокирован. Для того, чтобы воркер мог поддерживать блокировку и отключение процессора для обработки других запросов, необходимо использовать вспомогательное переключение сопрограмм, либо можно использовать и асинхронные клиенты, но асинхронные клиенты слишком вложены друг в друга, что очень неудобно. Сопрограмма swoole может добиться эффекта и производительности асинхронного ввода-вывода, незаметно написав синхронный код.
  • Второй — это канал, который реализует переключение сопрограмм и планирование на нижнем уровне.Ниже приводится подробное описание того, что такое канал.

3. Канал корутины/канала, аналогичныйgoлингвистическийchan, который поддерживает сопрограммы с несколькими производителями и сопрограммы с несколькими потребителями. Нижний слой автоматически реализует переключение и планирование сопрограмм. При высоком уровне параллелизма легко увидеть, что когда пул соединений пуст, если в качестве носителя для хранения переменной объекта соединения используется общий массив или splqueue(), это не может привести к эффекту блокировки в ожидании других запросов. выпущен, то есть он может напрямую возвращать только null. Итак, здесь мы используем очень мощный канал в сопрограмме swoole4 в качестве носителя данных через конвейер, а его метод удаления из очереди pop($timeout) может задавать блокировку и возвращаться через указанное время. Обратите внимание, что swoole2 — это параметр без тайм-аута, что не относится к этому сценарию. На языке го, если chan ждет или отправляет без потребления или производства один к одному, возникает взаимоблокировка. Таким образом, тайм-аут swoole4 должен быть сгенерирован, чтобы избежать бесконечного ожидания пустого канала. Основная ссылка:

wiki.Say Me Oh.com/wiki/Afraid/Afraid…

Пример переключения каналов:

<?php
use \Swoole\Coroutine\Channel;
$chan = new Channel();
go(function () use ($chan) {
    echo "我是第一个协程，等待3秒内有push就执行返回" . PHP_EOL;
    $p = $chan->pop(2);#1
    echo "pop返回结果" . PHP_EOL;
    var_dump($p);
});
go(function () use ($chan) {
    co::sleep(1);#2
    $chan->push(1);
});
echo "main" . PHP_EOL;

Код в #1 будет выполнен первым, а затем встретится с pop(), поскольку канал все еще пуст, он будет ждать 2 секунды. В это время сопрограмма освобождает процессор, переходит ко второй сопрограмме для выполнения, а затем № 2 переходит в спящий режим на 1 секунду.После того, как push-переменная 1 входит в канал, она возвращается к № 1 для продолжения выполнения и успешно извлекает автомобиль и передает только что введенное значение 1. Текущий результат:

Если вы замените время ожидания в #2 на время ожидания больше, чем pop(), результат будет следующим:

  • В соответствии с этими характеристиками абстрактный класс инкапсуляции, который окончательно реализует пул соединений:
<?php
/**
 * 连接池封装.
 * User: user
 * Date: 2018/9/1
 * Time: 13:36
 */

use Swoole\Coroutine\Channel;

abstract class AbstractPool
{
    private $min;//最少连接数
    private $max;//最大连接数
    private $count;//当前连接数
    private $connections;//连接池组
    protected $spareTime;//用于空闲连接回收判断
    //数据库配置
    protected $dbConfig = array(
        'host' => '10.0.2.2',
        'port' => 3306,
        'user' => 'root',
        'password' => 'root',
        'database' => 'test',
        'charset' => 'utf8',
        'timeout' => 2,
    );

    private $inited = false;

    protected abstract function createDb();

    public function __construct()
    {
        $this->min = 10;
        $this->max = 100;
        $this->spareTime = 10 * 3600;
        $this->connections = new Channel($this->max + 1);
    }

    protected function createObject()
    {
        $obj = null;
        $db = $this->createDb();
        if ($db) {
            $obj = [
                'last_used_time' => time(),
                'db' => $db,
            ];
        }
        return $obj;
    }

    /**
     * 初始换最小数量连接池
     * @return $this|null
     */
    public function init()
    {
        if ($this->inited) {
            return null;
        }
        for ($i = 0; $i < $this->min; $i++) {
            $obj = $this->createObject();
            $this->count++;
            $this->connections->push($obj);
        }
        return $this;
    }

    public function getConnection($timeOut = 3)
    {
        $obj = null;
        if ($this->connections->isEmpty()) {
            if ($this->count < $this->max) {//连接数没达到最大，新建连接入池
                $this->count++;
                $obj = $this->createObject();
            } else {
                $obj = $this->connections->pop($timeOut);//timeout为出队的最大的等待时间
            }
        } else {
            $obj = $this->connections->pop($timeOut);
        }
        return $obj;
    }

    public function free($obj)
    {
        if ($obj) {
            $this->connections->push($obj);
        }
    }

    /**
     * 处理空闲连接
     */
    public function gcSpareObject()
    {
        //大约2分钟检测一次连接
        swoole_timer_tick(120000, function () {
            $list = [];
            /*echo "开始检测回收空闲链接" . $this->connections->length() . PHP_EOL;*/
            if ($this->connections->length() < intval($this->max * 0.5)) {
                echo "请求连接数还比较多，暂不回收空闲连接\n";
            }#1
            while (true) {
                if (!$this->connections->isEmpty()) {
                    $obj = $this->connections->pop(0.001);
                    $last_used_time = $obj['last_used_time'];
                    if ($this->count > $this->min && (time() - $last_used_time > $this->spareTime)) {//回收
                        $this->count--;
                    } else {
                        array_push($list, $obj);
                    }
                } else {
                    break;
                }
            }
            foreach ($list as $item) {
                $this->connections->push($item);
            }
            unset($list);
        });
    }
}

  • Реализовано под синхронным клиентом PDO

<?php
/**
 * 数据库连接池PDO方式
 * User: user
 * Date: 2018/9/8
 * Time: 11:30
 */
require "AbstractPool.php";

class MysqlPoolPdo extends AbstractPool
{
    protected $dbConfig = array(
        'host' => 'mysql:host=10.0.2.2:3306;dbname=test',
        'port' => 3306,
        'user' => 'root',
        'password' => 'root',
        'database' => 'test',
        'charset' => 'utf8',
        'timeout' => 2,
    );
    public static $instance;

    public static function getInstance()
    {
        if (is_null(self::$instance)) {
            self::$instance = new MysqlPoolPdo();
        }
        return self::$instance;
    }

    protected function createDb()
    {
        return new PDO($this->dbConfig['host'], $this->dbConfig['user'], $this->dbConfig['password']);
    }
}

$httpServer = new swoole_http_server('0.0.0.0', 9501);
$httpServer->set(
    ['worker_num' => 1]
);
$httpServer->on("WorkerStart", function () {
    MysqlPoolPdo::getInstance()->init();
});
$httpServer->on("request", function ($request, $response) {
    $db = null;
    $obj = MysqlPoolPdo::getInstance()->getConnection();
    if (!empty($obj)) {
        $db = $obj ? $obj['db'] : null;
    }
    if ($db) {
        $db->query("select sleep(2)");
        $ret = $db->query("select * from guestbook limit 1");
        MysqlPoolPdo::getInstance()->free($obj);
        $response->end(json_encode($ret));
    }
});
$httpServer->start();

Детальный процесс вызова кода:
1. Когда сервер запускается, вызовите метод init () для инициализации минимального количества объектов подключения (указанные мин) и поместите их в объект соединений типа Type Channel. В вызове цикла в init он опирается на createObject (), чтобы вернуть объект подключения и createObject ()
В нем вызывается тот абстрактный метод, который был реализован изначально, а инициализация возвращает соединение PDO db. Итак, в настоящее время в соединениях пула соединений есть минимальные объекты.

2. Сервер прослушивает запросы пользователей, при приеме и отправке запросов вызывает метод getConnection() количества подключений к pop() объекту из канала подключений. В это время, если одновременно выполняются 10 запросов, сервер настроен с 1 рабочим процессом, поэтому, когда он переходит к объекту и возвращается, он сталкивается с запросом sleep(), поскольку используемый объект соединения является запросом pdo, и рабочий процесс в это время только и может дождаться завершения следующего запроса. Поэтому остальные соединения в пуле фактически избыточны, и скорость запросов синхронного клиента может быть связана только с количеством воркеров.
3. После завершения запроса вызовите метод free(), чтобы поместить объект соединения обратно в пул соединений.

Результат выполнения ab -c 10 -n 10, один рабочий процесс, выберите sleep(2) запрос sleep в течение 2 с, общее время работы в режиме синхронного клиента составляет более 20 с, а соединение mysql всегда поддерживается на один. Результат выглядит следующим образом:

  • Клиент Coroutine Вызов метода Coroutine\MySQL
<?php
/**
 * 数据库连接池协程方式
 * User: user
 * Date: 2018/9/8
 * Time: 11:30
 */
require "AbstractPool.php";

class MysqlPoolCoroutine extends AbstractPool
{
    protected $dbConfig = array(
        'host' => '10.0.2.2',
        'port' => 3306,
        'user' => 'root',
        'password' => 'root',
        'database' => 'test',
        'charset' => 'utf8',
        'timeout' => 10,
    );
    public static $instance;

    public static function getInstance()
    {
        if (is_null(self::$instance)) {
            self::$instance = new MysqlPoolCoroutine();
        }
        return self::$instance;
    }

    protected function createDb()
    {
        $db = new Swoole\Coroutine\Mysql();
        $db->connect(
            $this->dbConfig
        );
        return $db;
    }
}

$httpServer = new swoole_http_server('0.0.0.0', 9501);
$httpServer->set(
    ['worker_num' => 1]
);
$httpServer->on("WorkerStart", function () {
    //MysqlPoolCoroutine::getInstance()->init()->gcSpareObject();
    MysqlPoolCoroutine::getInstance()->init();
});

$httpServer->on("request", function ($request, $response) {
    $db = null;
    $obj = MysqlPoolCoroutine::getInstance()->getConnection();
    if (!empty($obj)) {
        $db = $obj ? $obj['db'] : null;
    }
    if ($db) {
        $db->query("select sleep(2)");
        $ret = $db->query("select * from guestbook limit 1");
        MysqlPoolCoroutine::getInstance()->free($obj);
        $response->end(json_encode($ret));
    }
});
$httpServer->start();

Подробное объяснение процесса вызова кода
1. Точно так же вызов в режиме клиента сопрограммы также реализует ранее инкапсулированный класс пула соединений AbstractPool.php. Просто абстрактный метод createDb() реализован с использованием встроенного в swoole клиента сопрограммы.
2. После запуска сервера инициализация аналогична синхронизации. Разница в том, что при получении объекта подключения, если в это время выполняется 10 одновременных запросов, рабочий процесс также настраивается на его обработку, но при поступлении первого запроса объект подключения в пуле выталкивается и выполняется для запроса ( ), когда сон заблокирован, в это время рабочий процесс не ждет завершения выбора, а переключается на другую сопрограмму для обработки следующего запроса. По завершении объект также выпускается в пул. Среди них getConnection() находится в сегменте кода, который в основном объясняется.

public function getConnection($timeOut = 3)
    {
        $obj = null;
        if ($this->connections->isEmpty()) {
            if ($this->count < $this->max) {//连接数没达到最大，新建连接入池
                $this->count++;
                $obj = $this->createObject();#1
            } else {
                $obj = $this->connections->pop($timeOut);#2 
            }
        } else {
            $obj = $this->connections->pop($timeOut);#3
        }
        return $obj;
    }

При вызове getConnection(), если соединения пула соединений пусты из-за большого количества одновременных запросов и максимальное число соединений не достигнуто, код переходит к #1, вызывает createObject() и возвращается новое соединение; если соединения пула соединений пусты и достигнуто максимальное количество соединений max, код выполняется до # 2, то есть $this->connections->pop($timeOut), что блокирует время $timeOut. ссылка освобождена, она будет успешно получена, а затем сопрограмма вернется. Если тайм-аут не получен, он вернет false.

3. Наконец, давайте поговорим о важной настройке сопрограммы клиента Mysql, которая представляет собой настройку значения тайм-аута в $dbConfig в коде. Эта конфигурация означает наибольшее время ожидания запроса. См. пример для иллюстрации:

go(function () {
    $start = microtime(true);
    $db = new Swoole\Coroutine\MySQL();
    $db->connect([
        'host' => '10.0.2.2',
        'port' => 3306,
        'user' => 'root',
        'password' => 'root',
        'database' => 'test',
        'timeout' => 4#1
    ]);
    $db->query("select sleep(5)");
    echo "我是第一个sleep五秒之后\n";
    $ret = $db->query("select user from guestbook limit 1");#2
    var_dump($ret);
    $use = microtime(true) - $start;
    echo "协程mysql输出用时：" . $use . PHP_EOL;
});

Код #1, если тайм-аут настроен на тайм-аут запроса 4 с, а первый запрос select sleep(5) заблокирован, сопрограмма переключается на выполнение следующего sql.На самом деле, $db не может быть успешно выполнено, потому что с соединение, то же самое. В сопрограмме выполнение фактически синхронно, поэтому в это время второй запрос не будет выполнен после ожидания 4-секундного тайм-аута и не получит соединение с базой данных для выполнения. Если время выполнения первого запроса меньше этого таймаута, запрос будет выполнен успешно. Угадайте, сколько времени ушло на выполнение вышеописанного? Результат выглядит следующим образом:

Если вы измените таймаут на 6 с, результат будет следующим:

Следует отметить, что реализация ассоциации в соответствии с клиентским процессом на самом деле синхронно, асинхронные не понимают, он только что встречается, когда IO позволяет блокировать реализацию права на переключение только на другой COROUTINE, и не следует путать асинхронно.

Результат выполнения ab -c 10 -n 10, один рабочий процесс, запрос select sleep(2) sleep на 2 секунды, общее время работы клиентского режима сопрограммы составляет более 2 секунд. Результат выглядит следующим образом:

Количество подключений к базе данных на данный момент равно 10 (показать полный PROCESSLIST):

Попробуйте снова ab -c 200 -n 1000 http://127.0.0.1:9501/, более 200 одновременных обработок, время больше 20 секунд, количество mysql соединений достигает указанного максимума 100. Результат выглядит следующим образом:

 

4. Послесловие

Теперь пул соединений в основном реализует распределение и управление соединениями с высокой степенью параллелизма, но есть еще некоторые детали, с которыми нужно разобраться, например:

  • При параллельной работе устанавливается максимальное количество объектов пула, и такое количество не может поддерживаться в пуле все время.Когда запрос простаивает, количество пулов соединений должно поддерживаться в пределах значения простоя. Вот простой метод gcSpareObject() для реализации обработки бездействия. Вызовите его непосредственно при инициализации пробуждающего устройства: MysqlPoolCoroutine::getInstance()->init()->gcSpareObject(); будет периодически обнаруживать и перезапускать. Вопрос в том, как определить, что программа относительно простаивает и заслуживает дальнейшей оптимизации.
  • Примерное обнаружение соединение живое, устранение мертвых цепей
  • Если программа забывает вызвать free() для выпуска объекта в пул, есть ли лучший способ избежать этого?

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

Категории