PDO Pool

Импорт почти готов: воркеры проверяют адреса через GeoDirectory. Осталось сохранять проверенные адреса в базу. Кажется, мелочь:

php

$pdo = new PDO('mysql:host=localhost;dbname=profiles', 'app', 'secret');

for ($i = 0; $i < 10; $i++) {
    $workers->spawn(function () use ($queue, $pdo) {
        foreach ($queue as $address) {
                checkAddress($address);
                saveAddress($pdo, $address);
        }
    });
}

Один объект PDO на десять корутин. В главе про каналы мы говорили, что гонок данных в одном потоке не бывает, значит, всё в порядке?

Нет. Для памяти это действительно так. Но соединение с базой не память, а протокол поверх сокета: запрос, ответ, запрос, ответ, строго по очереди. Каждый запрос к базе — это точка ожидания, в которой корутина засыпает, а управление переходит другой. И та пишет свой запрос в тот же самый сокет:

php

// Воркер 1
$pdo->beginTransaction();
$pdo->exec("INSERT INTO addresses ...");
// точка ожидания: воркер 1 заснул, проснулся воркер 2

// Воркер 2
$pdo->beginTransaction(); // на том же соединении!
$pdo->exec("UPDATE ...");
$pdo->commit(); // закоммитил и свою, и чужую транзакцию

Ответы перемешиваются, транзакции коммитят друг друга. Гонки вернулись, только теперь они живут не в памяти, а в соединении.

Хорошо, тогда каждой корутине по собственному соединению?

php

$workers->spawn(function () use ($queue) {
    $pdo = new PDO(/* ... */); // своё соединение
    // ...
});

Для десяти воркеров сойдёт. Но представьте не импорт, а сервер, где корутина создаётся на каждый запрос. Тысяча корутин — тысяча TCP-соединений. MySQL по умолчанию разрешает 151, PostgreSQL 100. Да и открывать соединение на несколько миллисекунд работы попросту дорого: рукопожатие с базой может занять больше, чем сам запрос.

Знакомая развилка, правда? В главе про каналы был тот же выбор: сто тысяч соединений к GeoDirectory или очередь на десятерых.

Пул: очередь наоборот

Решение называется пулом соединений: заранее открыть N соединений и выдавать их корутинам на время работы. Кстати, мы уже умеем такое строить. Пул — это канал, в котором лежат соединения:

php

$pool = new Channel(5);

for ($i = 0; $i < 5; $i++) {
    $pool->send(new PDO(/* ... */));
}

// В корутине:
$pdo = $pool->recv();   // взять соединение
saveAddress($pdo, $address);
$pool->send($pdo);      // вернуть на место

Свободные соединения лежат в буфере. Если все заняты, recv усыпит корутину до тех пор, пока кто-нибудь не вернёт соединение. Та же синхронизация из прошлой главы, только очередь стоит не из задач, а из ресурсов.

Схема рабочая, но у неё есть слабое место: вернуть соединение нужно не забыть, причём при любом исходе, включая исключение и отмену. А ещё транзакции, битые соединения, реконнекты. Для PDO всё это уже сделано за нас, прямо в ядре.

PDO Pool

Пул встроен в сам PDO и включается атрибутами конструктора:

php

$pdo = new PDO('mysql:host=localhost;dbname=profiles', 'app', 'secret', [
    PDO::ATTR_POOL_ENABLED => true,
    PDO::ATTR_POOL_MIN     => 2,
    PDO::ATTR_POOL_MAX     => 10,
]);

Снаружи ничего не изменилось: тот самый первый пример, где один $pdo уходит в десять воркеров, теперь корректен. Объект $pdo больше не соединение, а фасад пула. Когда корутина выполняет первый запрос, пул выдаёт ей собственное соединение, и все её запросы идут через него. Завершилась корутина — соединение вернулось в пул, к следующей корутине.

Никакого recv и send руками: взятие и возврат происходят сами, в правильные моменты, при любом исходе. Код выглядит как обычный синхронный PHP с обычным PDO, в этом и был замысел.

Транзакции

Транзакция — это состояние соединения, поэтому пул обращается с ней особо: пока транзакция открыта, соединение закреплено за корутиной и в пул не вернётся:

php

$workers->spawn(function () use ($pdo, $queue) {
    // ...
    $pdo->beginTransaction();
    $pdo->exec("INSERT INTO addresses (user_id, region) VALUES (...)");
    $pdo->exec("UPDATE users SET address_checked = 1 WHERE id = ...");
    $pdo->commit();
    // только теперь соединение может вернуться в пул
});

А если корутина завершилась, забыв про commit? Вспомните главу про Scope: воркер могли отменить прямо посреди транзакции, и это нормальный сценарий, а не катастрофа. Пул перед возвратом соединения автоматически выполнит ROLLBACK. Недописанная транзакция не утечёт следующей корутине и не повиснет в базе.

Когда соединение обрывается

Импорт идёт час. За час база может перезапуститься, сеть моргнуть, а DBA убить сессию. В классическом PHP скрипт бы просто упал, но долгоживущему приложению нужно уметь жить дальше.

Пул проверяет соединения при возврате: разорванное уничтожается, а не передаётся следующей корутине. А если запрос упал из-за обрыва, достаточно повторить его на том же $pdo, и пул выдаст корутине свежее соединение:

php

try {
    saveAddress($pdo, $address);
} catch (PDOException $e) {
    saveAddress($pdo, $address); // пул уже заменил соединение
}

Не нужно писать логику реконнекта, не нужно пересоздавать объект PDO. Достаточно повторить запрос, всё остальное пул сделает сам.

В итоге код работает так, как будто у каждой корутины есть собственное соединение с базой. На самом деле соединений всего десять, и пул постоянно передаёт их из рук в руки, следит за транзакциями и выбрасывает умершие. Но снаружи этой кухни не видно, и в этом главное достоинство: мы просто пишем обычный код с PDO.

Кстати, наши воркеры до сих пор работают вслепую: проверяют адреса, сохраняют их, но сколько адресов оказались битыми и какие именно, никто не считает. Как получить от корутин результаты и удобно их собрать? Этим и займёмся в следующей главе.

Редактировать страницу Спросить на форумеОбновлено: июль 2026 г.

PDO Pool ​

Пул: очередь наоборот ​

PDO Pool ​

Транзакции ​

Когда соединение обрывается ​

PDO Pool

Пул: очередь наоборот

PDO Pool

Транзакции

Когда соединение обрывается