PHP数据库连接池的简单实现

.gitignore

@@ -0,0 +1,16 @@
+nbproject/
+.idea/
+nbbuild/
+dist/
+nbdist/
+nbactions.xml
+nb-configuration.xml
+.nb-gradle/
+*.iml
+out
+gen
+*.cache
+*.log
+/config.inc.php
+build
+version

README.md

@@ -0,0 +1,14 @@
+# 在PHP中使用pdo驱动和gearman实现的数据库连接池
+
+[连接池实现参考](https://gonzalo123.com/2010/11/01/database-connection-pooling-with-php-and-gearman/)
+[gearman参考](https://blog.csdn.net/qq43599939/article/details/54177438)
+
+有待改进的点：
+1. 预先加载的连接数量要可配置
+2. 当实际的PDO连接数不够时刻自动增加
+3. 异常处理
+4. PDO方法支持
+5. ...
+
+
+

conf/PoolConf.php

@@ -0,0 +1,17 @@
+<?php
+class PoolConf
+{
+    const PG1 = 'PG1';
+    static $DB = array(
+        self::PG1 => array(
+            'dsn'      => "pgsql:dbname=gonzalo;host=localhost",
+            'username' => 'user',
+            'password' => 'password',
+            'options'  => null),
+    );
+
+    static $SERVERS = array(
+        array('127.0.0.1', 4730),
+        array('127.0.0.1', 4731),
+    );
+}

examples/client.php

@@ -0,0 +1,40 @@
+<?php
+include('../conf/PoolConf.php');
+include('../lib/Pool/Client.php');
+include('../lib/Pool/Server.php');
+include('../lib/Pool/Exception.php');
+include('../lib/Pool/Server/Connection.php');
+include('../lib/Pool/Server/Stmt.php');
+
+use Pool\Client;
+$conn = Client::singleton()->getConnection(PoolConf::PG1); // Server\Connection实例
+
+$sql = "SELECT * FROM TEST.TBL1";
+$stmt = $conn->prepare($sql); // Stmt实例
+
+$stmt->execute();
+$data = $stmt->fetchall();
+echo "<p>count: " . count($data) . "</p>";
+
+try {
+    $sql = "SELECT * FROM NON_EXISTENT_TABLE WHERE sealeccion=1";
+    $stmt = $conn->prepare($sql);
+
+    $stmt->execute();
+    $data = $stmt->fetchall();
+    echo "<p>count: " . count($data) . "</p>";
+
+} catch (Exception $e) {
+    echo "ERROR: " . $e->getMessage();
+}
+
+
+
+echo "<pre>";
+print_r(Client::singleton()->info(PoolConf::PG1));
+echo "</pre>";
+
+/*
+$conn->execute(array('NAME' => 'gonzalo'));
+$conn->commit();
+*/

examples/client2.php

@@ -0,0 +1,25 @@
+<?php
+include('../conf/PoolConf.php');
+include('../lib/Pool/Client.php');
+include('../lib/Pool/Server.php');
+include('../lib/Pool/Exception.php');
+include('../lib/Pool/Server/Connection.php');
+include('../lib/Pool/Server/Stmt.php');
+
+use Pool\Client;
+$conn = Client::singleton()->getConnection(PoolConf::PG1);
+
+$data = $conn->prepare("SELECT * FROM TEST.TBL1 WHERE FIELD1=:S")->execute(array('S' => 1))->fetchall();
+
+echo count($data);
+
+echo "<pre>";
+print_r(Client::singleton()->info(PoolConf::PG1));
+echo "</pre>";
+
+
+
+/*
+$conn->execute(array('NAME' => 'gonzalo'));
+$conn->commit();
+*/

examples/client3.php

@@ -0,0 +1,18 @@
+<?php
+include('../conf/PoolConf.php');
+include('../lib/Pool/Client.php');
+include('../lib/Pool/Server.php');
+include('../lib/Pool/Exception.php');
+include('../lib/Pool/Server/Connection.php');
+include('../lib/Pool/Server/Stmt.php');
+
+use Pool\Client;
+$conn = Client::singleton()->getConnection(PoolConf::PG1);
+
+$conn->beginTransaction();
+$data = $conn->prepare("SELECT * FROM TEST.TBL1 WHERE SELECCION=:S")->execute(array('S' => 1))->fetchall();
+$conn->rollback();
+
+echo "<pre>";
+print_r(Client::singleton()->info(PoolConf::PG1));
+echo "</pre>";

lib/Pool/Client.php

@@ -0,0 +1,52 @@
+<?php
+namespace Pool;
+
+class Client {
+
+    private static $_instance = null;
+
+    private $_client = null;
+
+    private static $_conn = array();
+
+    protected function __construct($client) {
+        $this->_client = $client;
+    }
+
+    public static function shutdown() {
+        if (count(self::$_conn) > 0) {
+            foreach (self::$_conn as $conn) {
+                $conn->release();
+            }
+        }
+
+    }
+
+    public static function singleton() {
+        if (is_null(self::$_instance)) {
+            register_shutdown_function(array("\Pool\Client", 'shutdown'));
+        }
+        $client = new \GearmanClient();
+        foreach (\PoolConf::$SERVERS as $server) {
+            $client->addServer($server[0], $server[1]);
+        }
+
+        self::$_instance = new Client($client);
+
+        return self::$_instance;
+
+    }
+
+    public function getConnection($key) {
+        $cid = $this->_client->do("getConnection", $key);
+        $conn = new Server\Connection($cid, $this->_client, $key);
+        self::$_conn[] = $conn;
+        return $conn;
+    }
+
+    public function info($key)
+    {
+        return unserialize($this->_client->do("info", $key));
+    }
+
+}

lib/Pool/Exception.php

@@ -0,0 +1,5 @@
+<?php
+namespace Pool;
+class Exception extends \Exception
+{
+}

lib/Pool/Server.php

@@ -0,0 +1,129 @@
+<?php
+namespace Pool;
+class Server {
+
+    private static $usedPool = array();
+
+    private static $pool = array();
+
+    private static $stmts = array();
+
+    private static $number = 10;
+
+    static function init()
+    {
+        self::resetAll();
+        // Create connections
+        foreach (array_keys(\PoolConf::$DB) as $key) {
+            // Create transaction pools
+            for($i=0; $i< self::$number; $i++) {
+                self::_newConnection($key);
+            }
+        }
+    }
+
+    private static function resetAll () {
+        self::$usedPool = array();
+        self::$stmts = array();
+        self::$pool = array();
+    }
+
+    public static function _newConnection ($key) {
+        $pdoConf = \PoolConf::$DB[$key];
+        $conn = new \PDO($pdoConf['dsn'], $pdoConf['username'], $pdoConf['password']);
+        $conn->setAttribute(\PDO::ATTR_ERRMODE, \PDO::ERRMODE_EXCEPTION);
+        self::$pool[$key][] = $conn;
+    }
+
+    public static function getConnection($key) {
+        $conn = array_shift(self::$pool[$key]);
+        $cid = uniqid();
+        self::$usedPool[$key][$cid] = $conn;
+
+        return $cid;
+
+    }
+
+    public static function prepare($key, $cid, $sql) {
+        $conn = Server::_getConnection($key, $cid);
+        $stmtId = '';
+        if (!is_null($conn)) {
+            $stmtId = md5(serialize(array($key, $sql)));
+            $stmt = Server::_getStmt($stmtId);
+            if (false === $stmt) {
+                $stmt = $conn->prepare($sql); // PDO实例
+            }
+            self::_setStmt($stmtId, $stmt);
+        } else {
+            echo "ERROR: cid:{$cid}, sql:{$sql}\n";
+        }
+        return $stmtId;
+
+    }
+
+    public static function execute($stmtId, $parameters) {
+        $stmt = Server::_getStmt($stmtId);
+        try {
+          $stmt->execute($parameters);
+        } catch (\PDOException $e) {
+            return serialize(new Exception($e->getMessage(), $e->getCode()));
+        }
+        self::_setStmt($stmtId, $stmt);
+        return $stmtId;
+    }
+
+    public static function fetchAll($stmtId) {
+        $stmt = Server::_getStmt($stmtId);
+        if (!is_null($stmt)) {
+            return $stmt->fetchAll();
+        } else {
+            echo "ERROR: stmtId:-{$stmtId}-\n";
+        }
+    }
+
+    public static function info($key) {
+        return array(
+            'usedPool' => count(self::$usedPool[$key]),
+            'pool'     => count(self::$pool[$key]),
+            'stmts'    => count(self::$stmts)
+        );
+
+    }
+
+    public static function release($key, $cid) {
+        $conn = array_shift(self::$usedPool[$key]);
+        self::$pool[$key][] = $conn;
+    }
+
+    public function beginTransaction($cid, $key) {
+        $conn = Server::_getConnection($key, $cid);
+        $conn->beginTransaction();
+    }
+
+    public function commit($cid, $key) {
+        $conn = Server::_getConnection($key, $cid);
+        $conn->commit();
+    }
+
+    public function rollback($cid, $key) {
+        $conn = Server::_getConnection($key, $cid);
+        $conn->rollback();
+    }
+
+    public static function _getConnection($key, $cid) {
+        return self::$usedPool[$key][$cid];
+    }
+
+    public static function _getStmt($stmtId) {
+        if (array_key_exists($stmtId, self::$stmts)) {
+            return self::$stmts[$stmtId];
+        } else {
+            return false;
+        }
+    }
+
+    public static function _setStmt($stmtId, $stmt) {
+        self::$stmts[$stmtId] = $stmt;
+    }
+
+}

lib/Pool/Server/Connection.php

@@ -0,0 +1,56 @@
+<?php
+namespace Pool\Server;
+
+class Connection {
+    private $_cid = null;
+    private $_client = null;
+    private $_key = null;
+
+    function __construct($cid, $client, $key)
+    {
+
+        $this->_cid    = $cid;
+        $this->_client = $client;
+        $this->_key    = $key;
+    }
+
+    public function prepare($sql) {
+        $stmtId = $this->_client->do('prepare', serialize(array(
+            'sql' => $sql,
+            'cid' => $this->_cid,
+            'key' => $this->_key,
+        )));
+
+        return new Stmt($stmtId, $this->_cid, $this->_client); // Stmt实例
+    }
+
+    public function release() {
+        $this->_client->do('release', serialize(array(
+            'key' => $this->_key,
+            'cid' => $this->_cid,
+        )));
+    }
+
+    public function beginTransaction() {
+        $this->_client->do('beginTransaction', serialize(array(
+            'cid' => $this->_cid,
+            'key' => $this->_key,
+        )));
+    }
+
+    public function commit() {
+        $this->_client->do('commit', serialize(array(
+            'cid' => $this->_cid,
+            'key' => $this->_key,
+        )));
+    }
+
+    public function rollback() {
+        $this->_client->do('rollback', serialize(array(
+            'cid' => $this->_cid,
+            'key' => $this->_key,
+        )));
+    }
+
+
+}