Merge pull request #148 from zhamao-robot/feature/mysql

完善 MySQL 连接池和组件

.github/workflows/integration-test.yml

@@ -27,7 +27,7 @@ jobs:
     strategy:
       matrix:
         operating-system: [ "ubuntu-latest", "macos-latest" ]
-        php-versions: [ "7.2", "7.3", "7.4", "8.0", "8.1" ]
+        php-versions: [ "7.4", "8.0", "8.1" ]
     steps:
       - name: Checkout code
         uses: actions/checkout@v2

composer.json

@@ -13,7 +13,7 @@
         }
     ],
     "require": {
-        "php": "^7.2 || ^7.3 || ^7.4 || ^8.0 || ^8.1",
+        "php": "^7.4 || ^8.0 || ^8.1",
         "ext-json": "*",
         "doctrine/dbal": "^2.13.1",
         "dragonmantank/cron-expression": "^3.3",
@@ -22,7 +22,7 @@
         "koriym/attributes": "^1.0",
         "onebot/libonebot": "dev-develop",
         "psr/container": "^2.0",
-        "symfony/console": "~6.0 || ~5.0 || ~4.0",
+        "symfony/console": "^6.0 || ^5.0 || ^4.0",
         "symfony/polyfill-ctype": "^1.19",
         "symfony/polyfill-mbstring": "^1.19",
         "symfony/polyfill-php80": "^1.16",

config/global.php

@@ -80,4 +80,18 @@ $config['file_server'] = [
     ],
 ];
 
+/* MySQL 数据库连接配置，框架将自动生成连接池，支持多个连接池 */
+$config['mysql'] = [
+    [
+        'pool_name' => '', // 默认只有一个空名称的连接池，如果需要多个连接池，请复制此段配置并修改参数和名称
+        'host' => '127.0.0.1', // 填写数据库服务器地址后才会创建数据库连接
+        'port' => 3306,
+        'username' => 'root',
+        'password' => 'ZhamaoTEST',
+        'dbname' => 'zm',
+        'charset' => 'utf8mb4',
+        'pool_size' => 64,
+    ],
+];
+
 return $config;

src/Globals/global_functions.php

@@ -12,6 +12,8 @@ use ZM\Container\ContainerInterface;
 use ZM\Context\Context;
 use ZM\Logger\ConsoleLogger;
 use ZM\Middleware\MiddlewareHandler;
+use ZM\Store\MySQL\MySQLException;
+use ZM\Store\MySQL\MySQLWrapper;
 
 // 防止重复引用引发报错
 if (function_exists('zm_internal_errcode')) {
@@ -165,3 +167,23 @@ function app(string $abstract = null, array $parameters = [])
 
     return resolve($abstract, $parameters);
 }
+
+/**
+ * 获取 MySQL 调用的类
+ *
+ * @throws MySQLException
+ */
+function mysql(string $name = '')
+{
+    return new MySQLWrapper($name);
+}
+
+/**
+ * 获取构建 MySQL 的类
+ *
+ * @throws MySQLException
+ */
+function mysql_builder(string $name = '')
+{
+    return (new MySQLWrapper($name))->createQueryBuilder();
+}

src/ZM/Command/BotCraft/BotCraftCommand.php

@@ -0,0 +1,22 @@
+<?php
+
+declare(strict_types=1);
+
+namespace ZM\Command\BotCraft;
+
+use Symfony\Component\Console\Command\Command;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Output\OutputInterface;
+
+/**
+ * TODO: 用于从命令行创建插件
+ */
+class BotCraftCommand extends Command
+{
+    protected static $defaultName = 'bc:make';
+
+    public function execute(InputInterface $input, OutputInterface $output)
+    {
+        return 0;
+    }
+}

src/ZM/Command/Server/ServerStartCommand.php

@@ -39,6 +39,7 @@ class ServerStartCommand extends ServerCommand
             new InputOption('disable-safe-exit', null, null, '关闭安全退出（关闭后按CtrlC时直接杀死进程）'),
             new InputOption('no-state-check', null, null, '关闭启动前框架运行状态检查'),
             new InputOption('private-mode', null, null, '启动时隐藏MOTD和敏感信息'),
+            new InputOption('print-process-pid', null, null, '打印所有进程的PID'),
         ]);
         $this->setDescription('Run zhamao-framework | 启动框架');
         $this->setHelp('直接运行可以启动');

src/ZM/ConsoleApplication.php

@@ -9,6 +9,7 @@ use Phar;
 use Symfony\Component\Console\Application;
 use Symfony\Component\Console\Input\InputInterface;
 use Symfony\Component\Console\Output\OutputInterface;
+use ZM\Command\BotCraft\BotCraftCommand;
 use ZM\Command\BuildCommand;
 use ZM\Command\CheckConfigCommand;
 use ZM\Command\Generate\SystemdGenerateCommand;
@@ -43,6 +44,7 @@ final class ConsoleApplication extends Application
         $this->add(new ServerStopCommand());        // server停止
         $this->add(new ServerStartCommand());       // 运行主服务的指令控制器
         $this->add(new SystemdGenerateCommand());   // 生成systemd文件
+        $this->add(new BotCraftCommand());          // 用于从命令行创建插件
         if (LOAD_MODE === 1) {                      // 如果是 Composer 模式加载的，那么可以输入 check:config 命令，检查配置文件是否需要更新
             $this->add(new CheckConfigCommand());
         }

src/ZM/Event/Listener/WorkerEventListener.php

@@ -11,10 +11,14 @@ use ZM\Annotation\AnnotationHandler;
 use ZM\Annotation\AnnotationMap;
 use ZM\Annotation\AnnotationParser;
 use ZM\Annotation\Framework\Init;
+use ZM\Config\ZMConfig;
 use ZM\Container\ContainerServicesProvider;
+use ZM\Exception\ConfigException;
 use ZM\Exception\ZMKnownException;
 use ZM\Framework;
 use ZM\Process\ProcessStateManager;
+use ZM\Store\MySQL\MySQLException;
+use ZM\Store\MySQL\MySQLPool;
 use ZM\Utils\ZMUtil;
 
 class WorkerEventListener
@@ -26,7 +30,7 @@ class WorkerEventListener
      *
      * @throws Throwable
      */
-    public function onWorkerStart()
+    public function onWorkerStart999()
     {
         // 自注册一下，刷新当前进程的logger进程banner
         ob_logger_register(ob_logger());
@@ -46,6 +50,18 @@ class WorkerEventListener
             ProcessStateManager::saveProcessState(ZM_PROCESS_WORKER, posix_getpid(), ['worker_id' => ProcessManager::getProcessId()]);
         }
 
+        // 打印进程ID
+        if (Framework::getInstance()->getArgv()['print-process-pid'] && ProcessManager::getProcessId() === 0) {
+            logger()->info("MASTER:\t" . ProcessStateManager::getProcessState(ZM_PROCESS_MASTER)['pid']);
+            if (ProcessStateManager::$process_mode['manager'] > 0) {
+                logger()->info("MANAGER:\t" . ProcessStateManager::getProcessState(ZM_PROCESS_MANAGER));
+            }
+        }
+        if (Framework::getInstance()->getArgv()['print-process-pid']) {
+            $i = ProcessManager::getProcessId();
+            logger()->info('WORKER#' . $i . ":\t" . ProcessStateManager::getProcessState(ZM_PROCESS_WORKER, $i));
+        }
+
         // 设置容器，注册容器提供商
         resolve(ContainerServicesProvider::class)->registerServices('global');
 
@@ -61,13 +77,14 @@ class WorkerEventListener
             Framework::getInstance()->stop();
         });
 
-        // TODO: 注册各种池子
+        // 注册各种池子
+        $this->initConnectionPool();
 
         // 加载用户代码资源
-        $this->loadUserSources();
+        $this->initUserPlugins();
 
         // handle @Init annotation
-        $this->handleInit();
+        $this->dispatchInit();
 
         // 回显 debug 日志：进程占用的内存
         $memory_total = memory_get_usage() / 1024 / 1024;
@@ -77,7 +94,7 @@ class WorkerEventListener
     /**
      * @throws ZMKnownException
      */
-    public function onWorkerStop()
+    public function onWorkerStop999()
     {
         logger()->debug('Worker #' . ProcessManager::getProcessId() . ' stopping');
         ProcessStateManager::removeProcessState(ZM_PROCESS_WORKER, ProcessManager::getProcessId());
@@ -87,7 +104,7 @@ class WorkerEventListener
      * 加载用户代码资源，包括普通插件、单文件插件、Composer 插件等
      * @throws Throwable
      */
-    private function loadUserSources()
+    private function initUserPlugins()
     {
         logger()->debug('Loading user sources');
 
@@ -114,7 +131,10 @@ class WorkerEventListener
         AnnotationMap::loadAnnotationByParser($parser);
     }
 
-    private function handleInit()
+    /**
+     * @throws Throwable
+     */
+    private function dispatchInit()
     {
         $handler = new AnnotationHandler(Init::class);
         $handler->setRuleCallback(function (Init $anno) {
@@ -122,4 +142,36 @@ class WorkerEventListener
         });
         $handler->handleAll();
     }
+
+    /**
+     * 初始化各种连接池
+     *
+     * TODO：未来新增其他db的连接池
+     *
+     * @throws ConfigException
+     * @throws MySQLException
+     */
+    private function initConnectionPool()
+    {
+        // 清空 MySQL 的连接池
+        foreach (MySQLPool::getAllPools() as $name => $pool) {
+            MySQLPool::destroyPool($name);
+        }
+
+        // 读取 MySQL 配置文件
+        $conf = ZMConfig::get('global.mysql');
+        if (is_array($conf) && !is_assoc_array($conf)) {
+            // 如果有多个数据库连接，则遍历
+            foreach ($conf as $conn_conf) {
+                if ($conn_conf['host'] !== '') {
+                    MySQLPool::create($conn_conf['pool_name'], $conn_conf);
+                }
+            }
+        } elseif (is_assoc_array($conf)) {
+            // 这种情况也支持，但是不推荐
+            if ($conf['host'] !== '') {
+                MySQLPool::create($conf['pool_name'], $conf);
+            }
+        }
+    }
 }

src/ZM/Framework.php

@@ -285,8 +285,8 @@ class Framework
 
         // 添加框架需要监听的顶层事件监听器
         // worker 事件
-        ob_event_provider()->addEventListener(WorkerStartEvent::getName(), [WorkerEventListener::getInstance(), 'onWorkerStart'], 999);
-        ob_event_provider()->addEventListener(WorkerStopEvent::getName(), [WorkerEventListener::getInstance(), 'onWorkerStop'], 999);
+        ob_event_provider()->addEventListener(WorkerStartEvent::getName(), [WorkerEventListener::getInstance(), 'onWorkerStart999'], 999);
+        ob_event_provider()->addEventListener(WorkerStopEvent::getName(), [WorkerEventListener::getInstance(), 'onWorkerStop999'], 999);
         // Http 事件
         ob_event_provider()->addEventListener(HttpRequestEvent::getName(), function () {
             global $starttime;

src/ZM/Middleware/MiddlewareHandler.php

@@ -75,8 +75,6 @@ class MiddlewareHandler
 
     public function getPipeClosure(callable $callback, $stack_id)
     {
-        unset($this->stack[$stack_id]);
-        /** @noinspection PhpUnnecessaryLocalVariableInspection */
         $pipe_func = function (array $mid_list, ...$args) use ($callback, $stack_id, &$pipe_func) {
             $return = true;
             try {
@@ -130,6 +128,7 @@ class MiddlewareHandler
             }
             return $result ?? $after_result ?? $exception_result ?? null;
         };
+        unset($this->stack[$stack_id]);
         return $pipe_func;
     }
 

src/ZM/Store/MySQL/MySQLConnection.php

@@ -0,0 +1,132 @@
+<?php
+
+/** @noinspection PhpComposerExtensionStubsInspection */
+
+declare(strict_types=1);
+
+namespace ZM\Store\MySQL;
+
+use Doctrine\DBAL\Driver\Connection;
+use Doctrine\DBAL\ParameterType;
+use PDO;
+use PDOException;
+use Swoole\Database\PDOProxy;
+
+class MySQLConnection implements Connection
+{
+    /** @var PDO|PDOProxy */
+    private $conn;
+
+    private $pool_name;
+
+    public function __construct($params)
+    {
+        logger()->debug('Constructing...');
+        $this->conn = MySQLPool::pool($params['dbName'])->get();
+        $this->pool_name = $params['dbName'];
+    }
+
+    public function __destruct()
+    {
+        logger()->debug('Destructing！！！');
+        MySQLPool::pool($this->pool_name)->put($this->conn);
+    }
+
+    /**
+     * @param  mixed          $sql
+     * @param  mixed          $options
+     * @throws MySQLException
+     */
+    public function prepare($sql, $options = [])
+    {
+        try {
+            logger()->debug('Running SQL prepare: ' . $sql);
+            $statement = $this->conn->prepare($sql, $options);
+            assert($statement !== false);
+        } catch (PDOException $exception) {
+            throw new MySQLException($exception->getMessage(), $exception->getCode(), $exception);
+        }
+        return new MySQLStatement($statement);
+    }
+
+    /**
+     * @throws MySQLException
+     */
+    public function query(...$args)
+    {
+        try {
+            $statement = $this->conn->query(...$args);
+            assert($statement !== false);
+        } catch (PDOException $exception) {
+            throw new MySQLException($exception->getMessage(), $exception->getCode(), $exception);
+        }
+        return new MySQLStatement($statement);
+    }
+
+    public function quote($value, $type = ParameterType::STRING)
+    {
+        return $this->conn->quote($value, $type);
+    }
+
+    /**
+     * @param  mixed          $sql
+     * @throws MySQLException
+     */
+    public function exec($sql)
+    {
+        try {
+            logger()->debug('Running SQL exec: ' . $sql);
+            $statement = $this->conn->exec($sql);
+            assert($statement !== false);
+            return $statement;
+        } catch (PDOException $exception) {
+            throw new MySQLException($exception->getMessage(), $exception->getCode(), $exception);
+        }
+    }
+
+    /**
+     * @param  null|mixed     $name
+     * @throws MySQLException
+     */
+    public function lastInsertId($name = null)
+    {
+        try {
+            return $name === null ? $this->conn->lastInsertId() : $this->conn->lastInsertId($name);
+        } catch (PDOException $exception) {
+            throw new MySQLException($exception->getMessage(), $exception->getCode(), $exception);
+        }
+    }
+
+    public function beginTransaction()
+    {
+        return $this->conn->beginTransaction();
+    }
+
+    public function commit()
+    {
+        return $this->conn->commit();
+    }
+
+    public function rollBack()
+    {
+        return $this->conn->rollBack();
+    }
+
+    public function errorCode()
+    {
+        return $this->conn->errorCode();
+    }
+
+    public function errorInfo()
+    {
+        return $this->conn->errorInfo();
+    }
+
+    /**
+     * @return mixed
+     */
+    public function getPoolName()
+    {
+        return $this->pool_name;
+    }
+}

src/ZM/Store/MySQL/MySQLDriver.php

@@ -0,0 +1,48 @@
+<?php
+
+declare(strict_types=1);
+
+namespace ZM\Store\MySQL;
+
+use Doctrine\DBAL\Driver as DoctrineDriver;
+use Doctrine\DBAL\Platforms\MySqlPlatform;
+use Doctrine\DBAL\Schema\MySqlSchemaManager;
+use ZM\Config\ZMConfig;
+
+class MySQLDriver implements DoctrineDriver
+{
+    public function connect(array $params, $username = null, $password = null, array $driverOptions = [])
+    {
+        logger()->debug('Requiring new connection');
+        return new MySQLConnection($params);
+    }
+
+    public function getDatabasePlatform(): MySqlPlatform
+    {
+        return new MySqlPlatform();
+    }
+
+    public function getSchemaManager($conn)
+    {
+        return new MySqlSchemaManager($conn);
+    }
+
+    public function getName()
+    {
+        return 'pdo_mysql_pool';
+    }
+
+    public function getDatabase($conn)
+    {
+        $conf = ZMConfig::get('global.mysql');
+
+        if ($conn instanceof MySQLConnection) {
+            foreach ($conf as $v) {
+                if (($v['name'] ?? $v['dbname']) === $conn->getPoolName()) {
+                    return $v['dbname'];
+                }
+            }
+        }
+        return '';
+    }
+}

src/ZM/Store/MySQL/MySQLException.php

@@ -0,0 +1,9 @@
+<?php
+
+declare(strict_types=1);
+
+namespace ZM\Store\MySQL;
+
+class MySQLException extends \ZM\Exception\ZMException
+{
+}

src/ZM/Store/MySQL/MySQLPool.php

@@ -0,0 +1,109 @@
+<?php
+
+/** @noinspection PhpComposerExtensionStubsInspection */
+
+declare(strict_types=1);
+
+namespace ZM\Store\MySQL;
+
+use OneBot\Driver\Driver;
+use OneBot\Driver\Interfaces\PoolInterface;
+use OneBot\Driver\Swoole\ObjectPool as SwooleObjectPool;
+use OneBot\Driver\Swoole\SwooleDriver;
+use OneBot\Driver\Workerman\ObjectPool as WorkermanObjectPool;
+use OneBot\Driver\Workerman\WorkermanDriver;
+use PDO;
+use RuntimeException;
+
+class MySQLPool
+{
+    /**
+     * @var array<string, SwooleObjectPool|WorkermanObjectPool> 连接池列表
+     */
+    private static $pools = [];
+
+    /**
+     * 通过配置文件创建一个 MySQL 连接池
+     *
+     * @throws MySQLException
+     */
+    public static function create(string $name, array $config)
+    {
+        $size = $config['pool_size'] ?? 128;
+        $connect_str = 'mysql:host={host};port={port};dbname={dbname};charset={charset}';
+        $table = [
+            '{host}' => $config['host'],
+            '{port}' => $config['port'],
+            '{dbname}' => $config['dbname'],
+            '{charset}' => $config['charset'] ?? 'utf8mb4',
+        ];
+        $connect_str = str_replace(array_keys($table), array_values($table), $connect_str);
+        self::checkExtension();
+        switch (Driver::getActiveDriverClass()) {
+            case WorkermanDriver::class:
+                self::$pools[$name] = new WorkermanObjectPool($size, PDO::class, $connect_str, $config['username'], $config['password'], $config['options'] ?? []);
+                break;
+            case SwooleDriver::class:
+                self::$pools[$name] = new SwooleObjectPool($size, PDO::class, $connect_str, $config['username'], $config['password'], $config['options'] ?? []);
+        }
+    }
+
+    /**
+     * 获取一个数据库连接池
+     *
+     * @param  string                               $name 连接池名称
+     * @return SwooleObjectPool|WorkermanObjectPool
+     */
+    public static function pool(string $name)
+    {
+        if (!isset(self::$pools[$name])) {
+            throw new RuntimeException("Pool {$name} not found");
+        }
+        return self::$pools[$name];
+    }
+
+    /**
+     * 获取所有数据库连接池
+     *
+     * @return PoolInterface[]
+     */
+    public static function getAllPools(): array
+    {
+        return self::$pools;
+    }
+
+    /**
+     * 销毁数据库连接池
+     *
+     * @param string $name 数据库连接池名称
+     */
+    public static function destroyPool(string $name)
+    {
+        unset(self::$pools[$name]);
+    }
+
+    /**
+     * 检查数据库启动必要的依赖扩展，如果不符合要求则抛出异常
+     *
+     * @throws MySQLException
+     */
+    public static function checkExtension()
+    {
+        ob_start();
+        phpinfo(); // 这个phpinfo是有用的，不能删除
+        $str = ob_get_clean();
+        $str = explode("\n", $str);
+        foreach ($str as $v) {
+            $v = trim($v);
+            if ($v == '') {
+                continue;
+            }
+            if (mb_strpos($v, 'API Extensions') === false) {
+                continue;
+            }
+            if (mb_strpos($v, 'pdo_mysql') === false) {
+                throw new MySQLException(zm_internal_errcode('E00028') . '未安装 mysqlnd php-mysql扩展。');
+            }
+        }
+    }
+}

src/ZM/Store/MySQL/MySQLQueryBuilder.php

@@ -0,0 +1,31 @@
+<?php
+
+declare(strict_types=1);
+
+namespace ZM\Store\MySQL;
+
+use Doctrine\DBAL\Query\QueryBuilder;
+use ZM\Store\MySQL\MySQLException as DbException;
+
+class MySQLQueryBuilder extends QueryBuilder
+{
+    private $wrapper;
+
+    public function __construct(MySQLWrapper $wrapper)
+    {
+        parent::__construct($wrapper->getConnection());
+        $this->wrapper = $wrapper;
+    }
+
+    /**
+     * @throws DbException
+     * @return int|MySQLStatementWrapper
+     */
+    public function execute()
+    {
+        if ($this->getType() === self::SELECT) {
+            return $this->wrapper->executeQuery($this->getSQL(), $this->getParameters(), $this->getParameterTypes());
+        }
+        return $this->wrapper->executeStatement($this->getSQL(), $this->getParameters(), $this->getParameterTypes());
+    }
+}

src/ZM/Store/MySQL/MySQLStatement.php

@@ -0,0 +1,123 @@
+<?php
+
+/**
+ * @noinspection PhpComposerExtensionStubsInspection
+ */
+
+declare(strict_types=1);
+
+namespace ZM\Store\MySQL;
+
+use Doctrine\DBAL\Driver\Statement;
+use Doctrine\DBAL\ParameterType;
+use IteratorAggregate;
+use PDO;
+use PDOStatement;
+use Traversable;
+
+class MySQLStatement implements IteratorAggregate, Statement
+{
+    /** @var PDOStatement */
+    private $statement;
+
+    public function __construct($obj)
+    {
+        $this->statement = $obj;
+    }
+
+    public function closeCursor()
+    {
+        return $this->statement->closeCursor();
+    }
+
+    public function columnCount()
+    {
+        return $this->statement->columnCount();
+    }
+
+    public function setFetchMode($fetchMode, $arg2 = null, $arg3 = [])
+    {
+        if ($arg2 !== null && $arg3 !== []) {
+            return $this->statement->setFetchMode($fetchMode, $arg2, $arg3);
+        }
+        if ($arg2 !== null && $arg3 === []) {
+            return $this->statement->setFetchMode($fetchMode, $arg2);
+        }
+        if ($arg2 === null && $arg3 !== []) {
+            return $this->statement->setFetchMode($fetchMode, $arg2, $arg3);
+        }
+
+        return $this->statement->setFetchMode($fetchMode);
+    }
+
+    public function fetch($fetchMode = PDO::FETCH_ASSOC, $cursorOrientation = PDO::FETCH_ORI_NEXT, $cursorOffset = 0)
+    {
+        return $this->statement->fetch($fetchMode, $cursorOrientation, $cursorOffset);
+    }
+
+    public function fetchAll($fetchMode = PDO::FETCH_ASSOC, $fetchArgument = null, $ctorArgs = null)
+    {
+        if ($fetchArgument === null && $ctorArgs === null) {
+            return $this->statement->fetchAll($fetchMode);
+        }
+        if ($fetchArgument !== null && $ctorArgs === null) {
+            return $this->statement->fetchAll($fetchMode, $fetchArgument);
+        }
+
+        return $this->statement->fetchAll($fetchMode, $fetchArgument, $ctorArgs);
+    }
+
+    public function fetchColumn($columnIndex = 0)
+    {
+        return $this->statement->fetchColumn($columnIndex);
+    }
+
+    public function bindValue($param, $value, $type = ParameterType::STRING)
+    {
+        return $this->statement->bindValue($param, $value, $type);
+    }
+
+    public function bindParam($param, &$variable, $type = ParameterType::STRING, $length = null)
+    {
+        return $this->statement->bindParam($param, $variable, $type, $length);
+    }
+
+    public function errorCode()
+    {
+        return $this->statement->errorCode();
+    }
+
+    public function errorInfo()
+    {
+        return $this->statement->errorInfo();
+    }
+
+    public function execute($params = null)
+    {
+        return $this->statement->execute($params);
+    }
+
+    public function rowCount()
+    {
+        return $this->statement->rowCount();
+    }
+
+    public function getIterator(): Traversable
+    {
+        while (($result = $this->statement->fetch()) !== false) {
+            yield $result;
+        }
+    }
+
+    /**
+     * @deprecated 最好不使用此方法，此方法可能存在 Bug
+     * @return mixed
+     */
+    public function current()
+    {
+        if (method_exists($this->statement, 'current')) {
+            return $this->statement->current();
+        }
+        return null;
+    }
+}

src/ZM/Store/MySQL/MySQLStatementWrapper.php

@@ -0,0 +1,239 @@
+<?php
+
+/**
+ * @noinspection PhpMissingReturnTypeInspection
+ * @noinspection PhpUnused
+ */
+
+declare(strict_types=1);
+
+namespace ZM\Store\MySQL;
+
+use Doctrine\DBAL\Driver\ResultStatement;
+use Doctrine\DBAL\ForwardCompatibility\Result;
+use Throwable;
+use Traversable;
+
+class MySQLStatementWrapper
+{
+    public $stmt;
+
+    public function __construct(?Result $stmt)
+    {
+        $this->stmt = $stmt;
+    }
+
+    /**
+     * 获取结果的迭代器
+     * wrapper method
+     * @return ResultStatement
+     */
+    public function getIterator()
+    {
+        return $this->stmt->getIterator();
+    }
+
+    /**
+     * 获取列数
+     * wrapper method
+     * @return int
+     */
+    public function columnCount()
+    {
+        return $this->stmt->columnCount();
+    }
+
+    /**
+     * wrapper method
+     * @throws MySQLException
+     * @return array|false|mixed
+     */
+    public function fetchNumeric()
+    {
+        try {
+            return $this->stmt->fetchNumeric();
+        } catch (Throwable $e) {
+            throw new MySQLException($e->getMessage(), $e->getCode(), $e);
+        }
+    }
+
+    /**
+     * wrapper method
+     * @throws MySQLException
+     * @return array|false|mixed
+     */
+    public function fetchAssociative()
+    {
+        try {
+            return $this->stmt->fetchAssociative();
+        } catch (Throwable $e) {
+            throw new MySQLException($e->getMessage(), $e->getCode(), $e);
+        }
+    }
+
+    /**
+     * wrapper method
+     * @throws MySQLException
+     * @return false|mixed
+     */
+    public function fetchOne()
+    {
+        try {
+            return $this->stmt->fetchOne();
+        } catch (Throwable $e) {
+            throw new MySQLException($e->getMessage(), $e->getCode(), $e);
+        }
+    }
+
+    /**
+     * wrapper method
+     * @throws MySQLException
+     */
+    public function fetchAllNumeric(): array
+    {
+        try {
+            return $this->stmt->fetchAllNumeric();
+        } catch (Throwable $e) {
+            throw new MySQLException($e->getMessage(), $e->getCode(), $e);
+        }
+    }
+
+    /**
+     * wrapper method
+     * @throws MySQLException
+     */
+    public function fetchAllAssociative(): array
+    {
+        try {
+            return $this->stmt->fetchAllAssociative();
+        } catch (Throwable $e) {
+            throw new MySQLException($e->getMessage(), $e->getCode(), $e);
+        }
+    }
+
+    /**
+     * wrapper method
+     * @throws MySQLException
+     */
+    public function fetchAllKeyValue(): array
+    {
+        try {
+            return $this->stmt->fetchAllKeyValue();
+        } catch (Throwable $e) {
+            throw new MySQLException($e->getMessage(), $e->getCode(), $e);
+        }
+    }
+
+    /**
+     * wrapper method
+     * @throws MySQLException
+     */
+    public function fetchAllAssociativeIndexed(): array
+    {
+        try {
+            return $this->stmt->fetchAllAssociativeIndexed();
+        } catch (Throwable $e) {
+            throw new MySQLException($e->getMessage(), $e->getCode(), $e);
+        }
+    }
+
+    /**
+     * wrapper method
+     * @throws MySQLException
+     */
+    public function fetchFirstColumn(): array
+    {
+        try {
+            return $this->stmt->fetchFirstColumn();
+        } catch (Throwable $e) {
+            throw new MySQLException($e->getMessage(), $e->getCode(), $e);
+        }
+    }
+
+    /**
+     * wrapper method
+     * @throws MySQLException
+     */
+    public function iterateNumeric(): Traversable
+    {
+        try {
+            return $this->stmt->iterateNumeric();
+        } catch (Throwable $e) {
+            throw new MySQLException($e->getMessage(), $e->getCode(), $e);
+        }
+    }
+
+    /**
+     * wrapper method
+     * @throws MySQLException
+     */
+    public function iterateAssociative(): Traversable
+    {
+        try {
+            return $this->stmt->iterateAssociative();
+        } catch (Throwable $e) {
+            throw new MySQLException($e->getMessage(), $e->getCode(), $e);
+        }
+    }
+
+    /**
+     * wrapper method
+     * @throws MySQLException
+     */
+    public function iterateKeyValue(): Traversable
+    {
+        try {
+            return $this->stmt->iterateKeyValue();
+        } catch (Throwable $e) {
+            throw new MySQLException($e->getMessage(), $e->getCode(), $e);
+        }
+    }
+
+    /**
+     * wrapper method
+     * @throws MySQLException
+     */
+    public function iterateAssociativeIndexed(): Traversable
+    {
+        try {
+            return $this->stmt->iterateAssociativeIndexed();
+        } catch (Throwable $e) {
+            throw new MySQLException($e->getMessage(), $e->getCode(), $e);
+        }
+    }
+
+    /**
+     * wrapper method
+     * @throws MySQLException
+     */
+    public function iterateColumn(): Traversable
+    {
+        try {
+            return $this->stmt->iterateColumn();
+        } catch (Throwable $e) {
+            throw new MySQLException($e->getMessage(), $e->getCode(), $e);
+        }
+    }
+
+    /**
+     * wrapper method
+     * @throws MySQLException
+     * @return int
+     */
+    public function rowCount()
+    {
+        try {
+            return $this->stmt->rowCount();
+        } catch (Throwable $e) {
+            throw new MySQLException($e->getMessage(), $e->getCode(), $e);
+        }
+    }
+
+    /**
+     * wrapper method
+     */
+    public function free(): void
+    {
+        $this->stmt->free();
+    }
+}

src/ZM/Store/MySQL/MySQLWrapper.php

@@ -0,0 +1,597 @@
+<?php
+
+declare(strict_types=1);
+
+/**
+ * @noinspection PhpUnused
+ */
+
+namespace ZM\Store\MySQL;
+
+use Closure;
+use Doctrine\DBAL\Cache\QueryCacheProfile;
+use Doctrine\DBAL\Connection;
+use Doctrine\DBAL\DriverManager;
+use Doctrine\DBAL\ParameterType;
+use Doctrine\DBAL\Types\Type;
+use Throwable;
+use Traversable;
+use ZM\Store\MySQL\MySQLException as DbException;
+
+class MySQLWrapper
+{
+    /** @var Connection */
+    private $connection;
+
+    /**
+     * MySQLWrapper constructor.
+     * @throws DbException
+     */
+    public function __construct(string $name)
+    {
+        try {
+            $this->connection = DriverManager::getConnection(['driverClass' => MySQLDriver::class, 'dbName' => $name]);
+        } catch (Throwable $e) {
+            throw new DbException($e->getMessage(), $e->getCode(), $e);
+        }
+    }
+
+    public function __destruct()
+    {
+        $this->connection->close();
+    }
+
+    /**
+     * wrapper method
+     */
+    public function getDatabase(): string
+    {
+        return $this->connection->getDatabase();
+    }
+
+    /**
+     * wrapper method
+     */
+    public function isAutoCommit(): bool
+    {
+        return $this->connection->isAutoCommit();
+    }
+
+    /**
+     * wrapper method
+     */
+    public function setAutoCommit(bool $auto_commit)
+    {
+        $this->connection->setAutoCommit($auto_commit);
+    }
+
+    /**
+     * wrapper method
+     * @throws DbException
+     * @return array|false
+     */
+    public function fetchAssociative(string $query, array $params = [], array $types = [])
+    {
+        try {
+            return $this->connection->fetchAssociative($query, $params, $types);
+        } catch (Throwable $e) {
+            throw new DbException($e->getMessage(), $e->getCode(), $e);
+        }
+    }
+
+    /**
+     * wrapper method
+     * @throws DbException
+     * @return array|false
+     */
+    public function fetchNumeric(string $query, array $params = [], array $types = [])
+    {
+        try {
+            return $this->connection->fetchNumeric($query, $params, $types);
+        } catch (Throwable $e) {
+            throw new DbException($e->getMessage(), $e->getCode(), $e);
+        }
+    }
+
+    /**
+     * @throws DbException
+     * @return false|mixed
+     */
+    public function fetchOne(string $query, array $params = [], array $types = [])
+    {
+        try {
+            return $this->connection->fetchOne($query, $params, $types);
+        } catch (Throwable $e) {
+            throw new DbException($e->getMessage(), $e->getCode(), $e);
+        }
+    }
+
+    /**
+     * wrapper method
+     */
+    public function isTransactionActive(): bool
+    {
+        return $this->connection->isTransactionActive();
+    }
+
+    /**
+     * @param  string      $table 表
+     * @throws DbException
+     */
+    public function delete(string $table, array $criteria, array $types = []): int
+    {
+        try {
+            return $this->connection->delete($table, $criteria, $types);
+        } catch (Throwable $e) {
+            throw new DbException($e->getMessage(), $e->getCode(), $e);
+        }
+    }
+
+    /**
+     * wrapper method
+     * @param int $level Sets the transaction isolation level
+     */
+    public function setTransactionIsolation(int $level): int
+    {
+        return $this->connection->setTransactionIsolation($level);
+    }
+
+    /**
+     * wrapper method
+     */
+    public function getTransactionIsolation(): ?int
+    {
+        return $this->connection->getTransactionIsolation();
+    }
+
+    /**
+     * wrapper method
+     * @param  string      $table 表名
+     * @throws DbException
+     */
+    public function update(string $table, array $data, array $criteria, array $types = []): int
+    {
+        try {
+            return $this->connection->update($table, $data, $criteria, $types);
+        } catch (Throwable $e) {
+            throw new DbException($e->getMessage(), $e->getCode(), $e);
+        }
+    }
+
+    /**
+     * wrapper method
+     * @param  string      $table 表名
+     * @throws DbException
+     */
+    public function insert(string $table, array $data, array $types = []): int
+    {
+        try {
+            return $this->connection->insert($table, $data, $types);
+        } catch (Throwable $e) {
+            throw new DbException($e->getMessage(), $e->getCode(), $e);
+        }
+    }
+
+    /**
+     * wrapper method
+     * @param string $str The name to be quoted
+     */
+    public function quoteIdentifier(string $str): string
+    {
+        return $this->connection->quoteIdentifier($str);
+    }
+
+    /**
+     * wrapper method
+     * @param mixed                $value
+     * @param null|int|string|Type $type
+     */
+    public function quote($value, $type = ParameterType::STRING)
+    {
+        return $this->connection->quote($value, $type);
+    }
+
+    /**
+     * wrapper method
+     * @param string                                                               $query  SQL query
+     * @param array<int, mixed>|array<string, mixed>                               $params Query parameters
+     * @param array<int, null|int|string|Type>|array<string, null|int|string|Type> $types  Parameter types
+     *
+     * @throws DbException
+     * @return array<int,array<int,mixed>>
+     */
+    public function fetchAllNumeric(string $query, array $params = [], array $types = []): array
+    {
+        try {
+            return $this->connection->fetchAllNumeric($query, $params, $types);
+        } catch (Throwable $e) {
+            throw new DbException($e->getMessage(), $e->getCode(), $e);
+        }
+    }
+
+    /**
+     * wrapper method
+     * @param string                                                               $query  SQL query
+     * @param array<int, mixed>|array<string, mixed>                               $params Query parameters
+     * @param array<int, null|int|string|Type>|array<string, null|int|string|Type> $types  Parameter types
+     *
+     * @throws DbException
+     * @return array<int,array<string,mixed>>
+     */
+    public function fetchAllAssociative(string $query, array $params = [], array $types = []): array
+    {
+        try {
+            return $this->connection->fetchAllAssociative($query, $params, $types);
+        } catch (Throwable $e) {
+            throw new DbException($e->getMessage(), $e->getCode(), $e);
+        }
+    }
+
+    /**
+     * wrapper method
+     * @param string                                           $query  SQL query
+     * @param array<int, mixed>|array<string, mixed>           $params Query parameters
+     * @param array<int, int|string>|array<string, int|string> $types  Parameter types
+     *
+     * @throws DbException
+     */
+    public function fetchAllKeyValue(string $query, array $params = [], array $types = []): array
+    {
+        try {
+            return $this->connection->fetchAllKeyValue($query, $params, $types);
+        } catch (Throwable $e) {
+            throw new DbException($e->getMessage(), $e->getCode(), $e);
+        }
+    }
+
+    /**
+     * wrapper method
+     * @param string                                           $query  SQL query
+     * @param array<int, mixed>|array<string, mixed>           $params Query parameters
+     * @param array<int, int|string>|array<string, int|string> $types  Parameter types
+     *
+     * @throws DbException
+     * @return array<mixed,array<string,mixed>>
+     */
+    public function fetchAllAssociativeIndexed(string $query, array $params = [], array $types = []): array
+    {
+        try {
+            return $this->connection->fetchAllAssociativeIndexed($query, $params, $types);
+        } catch (Throwable $e) {
+            throw new DbException($e->getMessage(), $e->getCode(), $e);
+        }
+    }
+
+    /**
+     * wrapper method
+     * @param string                                                               $query  SQL query
+     * @param array<int, mixed>|array<string, mixed>                               $params Query parameters
+     * @param array<int, null|int|string|Type>|array<string, null|int|string|Type> $types  Parameter types
+     *
+     * @throws DbException
+     * @return array<int,mixed>
+     */
+    public function fetchFirstColumn(string $query, array $params = [], array $types = []): array
+    {
+        try {
+            return $this->connection->fetchFirstColumn($query, $params, $types);
+        } catch (Throwable $e) {
+            throw new DbException($e->getMessage(), $e->getCode(), $e);
+        }
+    }
+
+    /**
+     * wrapper method
+     * @param string                                                               $query  SQL query
+     * @param array<int, mixed>|array<string, mixed>                               $params Query parameters
+     * @param array<int, null|int|string|Type>|array<string, null|int|string|Type> $types  Parameter types
+     *
+     * @throws DbException
+     * @return Traversable<int,array<int,mixed>>
+     */
+    public function iterateNumeric(string $query, array $params = [], array $types = []): Traversable
+    {
+        try {
+            return $this->connection->iterateNumeric($query, $params, $types);
+        } catch (Throwable $e) {
+            throw new DbException($e->getMessage(), $e->getCode(), $e);
+        }
+    }
+
+    /**
+     * wrapper method
+     * @param string                                                               $query  SQL query
+     * @param array<int, mixed>|array<string, mixed>                               $params Query parameters
+     * @param array<int, null|int|string|Type>|array<string, null|int|string|Type> $types  Parameter types
+     *
+     * @throws DbException
+     * @return Traversable<int,array<string,mixed>>
+     */
+    public function iterateAssociative(string $query, array $params = [], array $types = []): Traversable
+    {
+        try {
+            return $this->connection->iterateAssociative($query, $params, $types);
+        } catch (Throwable $e) {
+            throw new DbException($e->getMessage(), $e->getCode(), $e);
+        }
+    }
+
+    /**
+     * wrapper method
+     * @param string                                           $query  SQL query
+     * @param array<int, mixed>|array<string, mixed>           $params Query parameters
+     * @param array<int, int|string>|array<string, int|string> $types  Parameter types
+     *
+     * @throws DbException
+     * @return Traversable<mixed,mixed>
+     */
+    public function iterateKeyValue(string $query, array $params = [], array $types = []): Traversable
+    {
+        try {
+            return $this->connection->iterateKeyValue($query, $params, $types);
+        } catch (Throwable $e) {
+            throw new DbException($e->getMessage(), $e->getCode(), $e);
+        }
+    }
+
+    /**
+     * wrapper method
+     * @param string                                           $query  SQL query
+     * @param array<int, mixed>|array<string, mixed>           $params Query parameters
+     * @param array<int, int|string>|array<string, int|string> $types  Parameter types
+     *
+     * @throws DbException
+     * @return Traversable<mixed,array<string,mixed>>
+     */
+    public function iterateAssociativeIndexed(string $query, array $params = [], array $types = []): Traversable
+    {
+        try {
+            return $this->connection->iterateAssociativeIndexed($query, $params, $types);
+        } catch (Throwable $e) {
+            throw new DbException($e->getMessage(), $e->getCode(), $e);
+        }
+    }
+
+    /**
+     * wrapper method
+     * @param string                                                               $query  SQL query
+     * @param array<int, mixed>|array<string, mixed>                               $params Query parameters
+     * @param array<int, null|int|string|Type>|array<string, null|int|string|Type> $types  Parameter types
+     *
+     * @throws DbException
+     * @return Traversable<int,mixed>
+     */
+    public function iterateColumn(string $query, array $params = [], array $types = []): Traversable
+    {
+        try {
+            return $this->connection->iterateColumn($query, $params, $types);
+        } catch (Throwable $e) {
+            throw new DbException($e->getMessage(), $e->getCode(), $e);
+        }
+    }
+
+    /**
+     * wrapper method
+     * @param string                                                               $sql    SQL query
+     * @param array<int, mixed>|array<string, mixed>                               $params Query parameters
+     * @param array<int, null|int|string|Type>|array<string, null|int|string|Type> $types  Parameter types
+     *
+     * @throws DbException
+     */
+    public function executeQuery(string $sql, array $params = [], array $types = [], ?QueryCacheProfile $qcp = null): MySQLStatementWrapper
+    {
+        try {
+            $query = $this->connection->executeQuery($sql, $params, $types, $qcp);
+            return new MySQLStatementWrapper($query);
+        } catch (Throwable $e) {
+            throw new DbException($e->getMessage(), $e->getCode(), $e);
+        }
+    }
+
+    /**
+     * wrapper method
+     * @param  string                                                               $sql    SQL query
+     * @param  array<int, mixed>|array<string, mixed>                               $params Query parameters
+     * @param  array<int, null|int|string|Type>|array<string, null|int|string|Type> $types  Parameter types
+     * @throws DbException
+     */
+    public function executeCacheQuery(string $sql, array $params, array $types, QueryCacheProfile $qcp): MySQLStatementWrapper
+    {
+        try {
+            $query = $this->connection->executeCacheQuery($sql, $params, $types, $qcp);
+            return new MySQLStatementWrapper($query);
+        } catch (Throwable $e) {
+            throw new DbException($e->getMessage(), $e->getCode(), $e);
+        }
+    }
+
+    /**
+     * wrapper method
+     * @param string                                                               $sql    SQL statement
+     * @param array<int, mixed>|array<string, mixed>                               $params Statement parameters
+     * @param array<int, null|int|string|Type>|array<string, null|int|string|Type> $types  Parameter types
+     *
+     * @throws DbException
+     * @return int|string  the number of affected rows
+     */
+    public function executeStatement(string $sql, array $params = [], array $types = [])
+    {
+        try {
+            return $this->connection->executeStatement($sql, $params, $types);
+        } catch (Throwable $e) {
+            throw new DbException($e->getMessage(), $e->getCode(), $e);
+        }
+    }
+
+    /**
+     * wrapper method
+     */
+    public function getTransactionNestingLevel(): int
+    {
+        return $this->connection->getTransactionNestingLevel();
+    }
+
+    /**
+     * wrapper method
+     * @param  null|string      $name name of the sequence object from which the ID should be returned
+     * @return false|int|string a string representation of the last inserted ID
+     */
+    public function lastInsertId(?string $name = null)
+    {
+        return $this->connection->lastInsertId($name);
+    }
+
+    /**
+     * overwrite method to $this->connection->transactional()
+     * @throws DbException
+     * @return mixed
+     */
+    public function transactional(Closure $func)
+    {
+        $this->beginTransaction();
+        try {
+            $res = $func($this);
+            $this->commit();
+            return $res;
+        } catch (Throwable $e) {
+            $this->rollBack();
+            throw new DbException($e->getMessage(), $e->getCode(), $e);
+        }
+    }
+
+    /**
+     * wrapper method
+     * @throws DbException
+     */
+    public function setNestTransactionsWithSavepoints(bool $nest_transactions_with_savepoints)
+    {
+        try {
+            $this->connection->setNestTransactionsWithSavepoints($nest_transactions_with_savepoints);
+        } catch (Throwable $e) {
+            throw new DbException($e->getMessage(), $e->getCode(), $e);
+        }
+    }
+
+    /**
+     * wrapper method
+     */
+    public function getNestTransactionsWithSavepoints(): bool
+    {
+        return $this->connection->getNestTransactionsWithSavepoints();
+    }
+
+    /**
+     * wrapper method
+     */
+    public function beginTransaction(): bool
+    {
+        return $this->connection->beginTransaction();
+    }
+
+    /**
+     * wrapper method
+     * @throws DbException
+     */
+    public function commit(): bool
+    {
+        try {
+            return $this->connection->commit();
+        } catch (Throwable $e) {
+            throw new DbException($e->getMessage(), $e->getCode(), $e);
+        }
+    }
+
+    /**
+     * wrapper method
+     * @throws DbException
+     */
+    public function rollBack(): bool
+    {
+        try {
+            return $this->connection->rollBack();
+        } catch (Throwable $e) {
+            throw new DbException($e->getMessage(), $e->getCode(), $e);
+        }
+    }
+
+    /**
+     * wrapper method
+     * @param  string      $savepoint the name of the savepoint to create
+     * @throws DbException
+     */
+    public function createSavepoint(string $savepoint)
+    {
+        try {
+            $this->connection->createSavepoint($savepoint);
+        } catch (Throwable $e) {
+            throw new DbException($e->getMessage(), $e->getCode(), $e);
+        }
+    }
+
+    /**
+     * wrapper method
+     * @param  string      $savepoint the name of the savepoint to release
+     * @throws DbException
+     */
+    public function releaseSavepoint(string $savepoint)
+    {
+        try {
+            $this->connection->releaseSavepoint($savepoint);
+        } catch (Throwable $e) {
+            throw new DbException($e->getMessage(), $e->getCode(), $e);
+        }
+    }
+
+    /**
+     * wrapper method
+     * @param  string      $savepoint the name of the savepoint to rollback to
+     * @throws DbException
+     */
+    public function rollbackSavepoint(string $savepoint)
+    {
+        try {
+            $this->connection->rollbackSavepoint($savepoint);
+        } catch (Throwable $e) {
+            throw new DbException($e->getMessage(), $e->getCode(), $e);
+        }
+    }
+
+    /**
+     * wrapper method
+     * @throws DbException
+     */
+    public function setRollbackOnly()
+    {
+        try {
+            $this->connection->setRollbackOnly();
+        } catch (Throwable $e) {
+            throw new DbException($e->getMessage(), $e->getCode(), $e);
+        }
+    }
+
+    /**
+     * wrapper method
+     * @throws DbException
+     */
+    public function isRollbackOnly(): bool
+    {
+        try {
+            return $this->connection->isRollbackOnly();
+        } catch (Throwable $e) {
+            throw new DbException($e->getMessage(), $e->getCode(), $e);
+        }
+    }
+
+    /**
+     * overwrite method to $this->connection->createQueryBuilder
+     */
+    public function createQueryBuilder(): MySQLQueryBuilder
+    {
+        return new MySQLQueryBuilder($this);
+    }
+
+    public function getConnection(): Connection
+    {
+        return $this->connection;
+    }
+}

tests/ZM/Utils/ZMUtilTest.php

@@ -1,10 +1,16 @@
 <?php
 
+declare(strict_types=1);
+
 namespace Tests\ZM\Utils;
 
+use PHPUnit\Framework\TestCase;
 use ZM\Utils\ZMUtil;
 
-class ZMUtilTest extends \PHPUnit\Framework\TestCase
+/**
+ * @internal
+ */
+class ZMUtilTest extends TestCase
 {
     public function testComposer()
     {