不知道大家有沒有遇到因為耗時任務(wù)，連接池耗盡，無法處理后續(xù)請求的問題。
因為這個問題也是苦惱，文檔雖然提供了一些連接池的案例，但是好像和業(yè)務(wù)不太匹配。也無法使用laravel orm。
所以自己寫了一個連接池用例。
支持laravel orm操作
支持手動取還，快取快還，解決連接被占用，盡快釋放出來給其他請求使用。

<?php

namespace app\tools;

use Illuminate\Container\Container;
use Illuminate\Database\Connection;
use Illuminate\Database\Connectors\ConnectionFactory;
use Swoole\Coroutine\Channel;
use Swoole\Timer;

class MysqlPool
{
    protected static $instance = []; //不同配置不同連接池實例
    protected static $lock;
    public static function getInstance($connectionName = "mysql"): self
    {
        if (!isset(self::$instance[$connectionName])) {
            if (self::$lock === null) {
                self::$lock = new Channel(1); // 容量為1的鎖通道
            }
            self::$lock->push(true); // 加鎖
            if (!isset(self::$instance[$connectionName])) {
                self::$instance[$connectionName] = new self($connectionName);
                self::$lock->close(); //關(guān)閉鎖通道
            }
        }
        return self::$instance[$connectionName];
    }

    public static function get($connectionName = "mysql")
    {
        return self::getInstance($connectionName)->getConnection();
    }

    public static function put(Connection $connection)
    {
        self::getInstance($connection->getName())->releaseConnection($connection);
    }

    protected $pool;
    protected $config;
    protected $connectionName;
    protected $wait_timeout;
    protected $heartbeat_interval;
    protected $maxConnections;
    private function __construct($connectionName)
    {
        $this->connectionName = $connectionName;
        $this->config = config("database.connections.{$connectionName}");
        $this->maxConnections = $this->config['pool']['max_connections'] ?? 5;
        $this->wait_timeout = $this->config['pool']['wait_timeout'] ?? 3;
        $this->heartbeat_interval = $this->config['pool']['heartbeat_interval'] ?? 30;
        $this->pool = new Channel($this->maxConnections);
        $this->initPool();
    }
    protected function initPool(): void
    {
        for ($i = 0; $i < $this->maxConnections; $i++) {
            $this->pool->push($this->createConnection());
        }
        //異步心跳檢測，重連檢測
        Timer::tick($this->heartbeat_interval * 1000, function () {
            $pool = [];
            while (!$this->pool->isEmpty()) {
                $conn = $this->pool->pop(0.1);
                try {
                    $conn->statement('SELECT 1');
                    $pool[] = $conn;
                } catch (\Exception $e) {
                    $pool[] = $this->createConnection();
                }
            }

            for ($i = 0; $i < $this->maxConnections; $i++) {
                if ($i < $this->maxConnections) {
                    //中斷導致沒有歸還，進行填充
                    $this->pool->push($pool[$i] ?? $this->createConnection());
                } else {
                    //釋放多余連接
                    $pool[$i]->disconnect();
                }
            }
        });
    }
    protected function createConnection(): Connection
    {
        $container = new Container(); //起一個新的容器服務(wù)，防止單例復用
        $factory = new ConnectionFactory($container); //該工廠是根據(jù)容器判斷是否使用單例
        $DB = $factory->make($this->config, $this->connectionName); //連接配置
        return $DB;
    }

    public function getConnection(): Connection
    {
        //提高容錯，防止其他代碼中斷導致連接沒有歸還，新的連接沒有連接可用
        return $this->pool->pop($this->wait_timeout) ?: $this->createConnection();
    }

    public function releaseConnection(Connection $connection): void
    {
        if ($connection->getPdo()->inTransaction()) {
            $connection->rollBack();
        }
        $this->pool->push($connection);
    }
}

//使用方式
$Db = MysqlPool::get(); //獲取一個連接實例
$info = $Db->table("xxx")->where([
    ['id', '=', $item['xx']],
    ['name', '=', $item['xx']],
])->first();//orm操作
MysqlPool::put($Db);//歸還連接實例

<?php

namespace app\tools;

use Swoole\Coroutine\Channel;
use Swoole\Timer;

class RedisPool
{
    protected static $instance = []; //不同配置不同連接池實例
    protected static $lock;
    private static function getInstance($connectionName = "default"): self
    {

        if (!isset(self::$instance[$connectionName])) {
            if (self::$lock === null) {
                self::$lock = new Channel(1); // 容量為1的鎖通道
            }
            self::$lock->push(true); // 加鎖
            if (!isset(self::$instance[$connectionName])) {
                self::$instance[$connectionName] = new self($connectionName);
                self::$lock->close(); //關(guān)閉鎖通道
            }
        }
        return self::$instance[$connectionName];
    }

    public static function get($connectionName = "default")
    {
        return self::getInstance($connectionName)->getConnection();
    }

    public static function put(\Redis $connection, $connectionName = "default")
    {
        self::getInstance($connectionName)->releaseConnection($connection);
    }

    protected $pool;
    protected $config;
    protected $connectionName;
    protected $wait_timeout;
    protected $heartbeat_interval;
    protected $maxConnections;
    private function __construct($connectionName)
    {
        $this->connectionName = $connectionName;
        $this->config = config("redis.{$connectionName}");
        $this->maxConnections = $this->config['pool']['max_connections'] ?? 5;
        $this->wait_timeout = $this->config['pool']['wait_timeout'] ?? 3;
        $this->heartbeat_interval = $this->config['pool']['heartbeat_interval'] ?? 30;
        $this->pool = new Channel($this->maxConnections);
        $this->initPool();
    }

    protected function initPool(): void
    {
        for ($i = 0; $i < $this->maxConnections; $i++) {
            $this->pool->push($this->createConnection());
        }
        Timer::tick($this->heartbeat_interval * 1000, function () {
            $pool = [];
            while (!$this->pool->isEmpty()) {
                $conn = $this->pool->pop(0.1);
                try {
                    $conn->ping();
                    $pool[] = $conn;
                } catch (\Exception $e) {
                    $pool[] = $this->createConnection();
                }
            }
            for ($i = 0; $i < $this->maxConnections; $i++) {
                if ($i < $this->maxConnections) {
                    //中斷導致沒有歸還，進行填充
                    $this->pool->push($pool[$i] ?? $this->createConnection());
                } else {
                    //釋放多余連接
                    $pool[$i]->close();
                }
            }
        });
    }
    protected function createConnection(): \Redis
    {
        $redis  = new \Redis();
        $redis->connect($this->config['host'], $this->config['port']);
        $this->config['password'] && $redis->auth($this->config['password']);
        $redis->select($this->config['database'] ?? 0);
        $redis->setOption(\Redis::OPT_PREFIX, $this->config['prefix'] ?? "");
        return $redis;
    }
    public function getConnection(): \Redis
    {
        //提高容錯，防止其他代碼中斷導致連接沒有歸還，新的連接沒有連接可用
        return $this->pool->pop($this->wait_timeout) ?: $this->createConnection();
    }

    public function releaseConnection(\Redis $redis): void
    {
        $redis->discard();
        $this->pool->push($redis);
    }
}
//使用方式
 $redis = RedisPool::get();//取一個連接實例
 $add_limiter = $redis->get("xxxx");
 RedisPool::put($redis);//歸還連接