Pool コネクションプール
複数のコルーチンが同じコネクションを共有するとデータの混乱を引き起こす可能性があるため、データベースや Redis などの接続リソースを管理するためにコネクションプールを使用する必要があります。
ヒント
この機能は workerman>=5.1.0 が必要です。
注意
- 下層は自動的に Swoole/Swow/Fiber/Select/Event ドリブンをサポートします。
- Fiber/Select/Event ドリブンを使用する場合、PDO Redis などのブロッキング拡張を使用していると、接続プールは自動的に 1 つのコネクションのみのプールに退化します。
Redisコネクションプール
<?php
use Workerman\Connection\TcpConnection;
use Workerman\Coroutine\Pool;
use Workerman\Events\Swoole;
use Workerman\Protocols\Http\Request;
use Workerman\Worker;
require_once __DIR__ . '/vendor/autoload.php';
class RedisPool
{
private Pool $pool;
public function __construct($host, $port, $max_connections = 10)
{
$this->pool = new Pool($max_connections);
// コネクションプールがコネクションを作成するためのメソッドを設定
$this->pool->setConnectionCreator(function () use ($host, $port) {
$redis = new \Redis();
$redis->connect($host, $port);
return $redis;
});
// コネクションプールがコネクションを破棄するためのメソッドを設定
$this->pool->setConnectionCloser(function ($redis) {
$redis->close();
});
// ハートビートチェックメソッドを設定
$this->pool->setHeartbeatChecker(function ($redis) {
$redis->ping();
});
}
// コネクションを取得
public function get(): \Redis
{
return $this->pool->get();
}
// コネクションを返却
public function put($redis): void
{
$this->pool->put($redis);
}
}
// Http サーバー
$worker = new Worker('http://0.0.0.0:8001');
$worker->eventLoop = Swoole::class; // または Swow::class または Fiber::class
$worker->onMessage = function (TcpConnection $connection, Request $request) {
static $pool;
if (!$pool) {
$pool = new RedisPool('127.0.0.1', 6379, 10);
}
$redis = $pool->get();
$redis->set('key', 'hello');
$value = $redis->get('key');
$pool->put($redis);
$connection->send($value);
};
Worker::runAll();
MySQLコネクションプール(自動的にコネクションを取得して返却をサポート)
<?php
use Workerman\Connection\TcpConnection;
use Workerman\Coroutine\Context;
use Workerman\Coroutine;
use Workerman\Coroutine\Pool;
use Workerman\Events\Swoole;
use Workerman\Protocols\Http\Request;
use Workerman\Worker;
require_once __DIR__ . '/vendor/autoload.php';
class Db
{
private static ?Pool $pool = null;
public static function __callStatic($name, $arguments)
{
if (self::$pool === null) {
self::initializePool();
}
// コルーチンコンテキストからコネクションを取得し、同じコルーチンが同じコネクションを使用することを保証
$pdo = Context::get('pdo');
if (!$pdo) {
// コネクションプールからコネクションを取得
$pdo = self::$pool->get();
Context::set('pdo', $pdo);
// コルーチンが終了するとき、自動的にコネクションを返却
Coroutine::defer(function () use ($pdo) {
self::$pool->put($pdo);
});
}
return call_user_func_array([$pdo, $name], $arguments);
}
private static function initializePool(): void
{
self::$pool = new Pool(10);
self::$pool->setConnectionCreator(function () {
return new \PDO('mysql:host=127.0.0.1;dbname=your_database', 'your_username', 'your_password');
});
self::$pool->setConnectionCloser(function ($pdo) {
$pdo = null;
});
self::$pool->setHeartbeatChecker(function ($pdo) {
$pdo->query('SELECT 1');
});
}
}
// Http サーバー
$worker = new Worker('http://0.0.0.0:8001');
$worker->eventLoop = Swoole::class; // または Swow::class または Fiber::class
$worker->onMessage = function (TcpConnection $connection, Request $request) {
$value = Db::query('SELECT NOW() as now')->fetchAll();
$connection->send(json_encode($value));
};
Worker::runAll();
インターフェース説明
interface PoolInterface
{
/**
* コンストラクタ
* @param int $max_connections 最大接続数、デフォルトは1
* @param array $config = [
* 'min_connections' => 1, // 最小接続数、デフォルトは1
* 'idle_timeout' => 60, // 接続のアイドルタイムアウト(秒)、デフォルトは60秒、60秒を超えると接続は破棄されて接続プールから削除される
* 'heartbeat_interval' => 50, // ハートビートチェックの間隔(秒)、デフォルトは50秒、50秒ごとに接続が正常かどうかチェックされる
* 'wait_timeout' => 10, // コネクション取得の待機タイムアウト(秒)、デフォルトは10秒、10秒を超えるとコネクション取得に失敗し例外がスローされる
* ]
*/
public function __construct(int $max_connections = 1, array $config = []);
/**
* コネクションを取得
*/
public function get(): mixed;
/**
* コネクションを返却
*/
public function put(object $connection): void;
/**
* コネクションを作成
*/
public function createConnection(): object;
/**
* コネクションを閉じ、接続プールから削除
*/
public function closeConnection(object $connection): void;
/**
* コネクションプール内の現在のコネクション数を取得(使用中のコネクションと未使用のコネクションを含む)
*/
public function getConnectionCount(): int;
/**
* 接続プール内のコネクションを閉じる(使用中のコネクションは含まない)
*/
public function closeConnections(): void;
/**
* コネクションプールがコネクションを作成するためのメソッドを設定
*/
public function setConnectionCreator(callable $connectionCreateHandler): self;
/**
* コネクションプールがコネクションを破棄するためのメソッドを設定
*/
public function setConnectionCloser(callable $connectionDestroyHandler): self;
/**
* ハートビートチェックメソッドを設定
*/
public function setHeartbeatChecker(callable $connectionHeartbeatHandler): self;
}