RPC 连接池自动保活与动态扩缩容

大白话先说清楚

没有连接池：
  每个请求 → 新建TCP连接 → 用完关掉
  1000并发 → 建1000次连接 → 慢、耗资源

有了连接池：
  启动时建10个连接放池子里
  请求来了 → 从池子取一个 → 用完放回去
  池子空了且并发高 → 自动多建几个
  闲了 → 超时没人用的连接自动关掉

Channel是什么：
  就是一个带锁的队列
  放连接：push
  取连接：pop（没有就等）
  天然支持协程并发，不用自己加锁

连接池生命周期：

启动 → 建minSize个连接放进Channel
    │
    ├─ 有请求 → pop取连接 → 用 → push还回去
    │
    ├─ 池子空了 → 当前连接数 < maxSize → 新建连接
    │
    ├─ 池子空了 → 当前连接数 == maxSize → 等待（阻塞）
    │
    └─ 定时检查 → 连接空闲太久 → 关掉回收

代码

<?php
// ConnectionPool.php

class RpcConnection
{
    public float $lastUsed;   // 最后使用时间
    public bool  $healthy = true;

    private Swoole\Coroutine\Client $client;

    public function __construct(private string $host, private int $port)
    {
        $this->client   = new Swoole\Coroutine\Client(SWOOLE_SOCK_TCP);
        $this->lastUsed = microtime(true);
        $this->connect();
    }

    private function connect(): void
    {
        if (!$this->client->connect($this->host, $this->port, 3.0)) {
            $this->healthy = false;
            throw new \RuntimeException("连接失败 {$this->host}:{$this->port}");
        }
        echo "新连接建立 {$this->host}:{$this->port}\n";
    }

    // 发请求
    public function send(string $data): string
    {
        // 连接断了就重连
        if (!$this->client->isConnected()) {
            $this->connect();
        }

        $this->client->send($data);
        $result         = $this->client->recv(5.0); // 5秒超时
        $this->lastUsed = microtime(true);

        if ($result === false) {
            $this->healthy = false;
            throw new \RuntimeException('连接异常，数据收取失败');
        }
        return $result;
    }

    public function close(): void
    {
        $this->client->close();
        echo "连接已关闭 {$this->host}:{$this->port}\n";
    }

    public function isConnected(): bool
    {
        return $this->client->isConnected();
    }
}

<?php
// Pool.php
require __DIR__ . '/ConnectionPool.php';

class ConnectionPool
{
    private Swoole\Coroutine\Channel $channel; // 连接队列
    private int $currentSize = 0;              // 当前连接总数

    public function __construct(
        private string $host,
        private int    $port,
        private int    $minSize    = 5,    // 最少保持5个连接
        private int    $maxSize    = 50,   // 最多建50个连接
        private int    $idleTime   = 60,   // 空闲超过60秒回收
        private float  $waitTime   = 3.0,  // 等不到连接最多等3秒
    ) {
        $this->channel = new Swoole\Coroutine\Channel($maxSize);
        $this->init();
        $this->startIdleChecker();
    }

    // 启动时预建minSize个连接
    private function init(): void
    {
        for ($i = 0; $i < $this->minSize; $i++) {
            $conn = $this->makeConnection();
            if ($conn) $this->channel->push($conn);
        }
        echo "连接池初始化完成，当前连接数：{$this->currentSize}\n";
    }

    // 取连接
    public function get(): RpcConnection
    {
        // 池子有空闲连接，直接取
        if (!$this->channel->isEmpty()) {
            $conn = $this->channel->pop(0.001);
            if ($conn && $conn->isConnected()) return $conn;
            // 连接断了，扔掉重建一个
            $this->currentSize--;
        }

        // 池子空了，还没到上限，新建连接
        if ($this->currentSize < $this->maxSize) {
            $conn = $this->makeConnection();
            if ($conn) {
                echo "并发上升，扩容至 {$this->currentSize} 个连接\n";
                return $conn; // 新建的直接返回，不先入队
            }
        }

        // 已到上限，等别人还回来
        echo "连接池满，等待中...\n";
        $conn = $this->channel->pop($this->waitTime);
        if (!$conn) throw new \RuntimeException("等待连接超时 {$this->waitTime}s");

        return $conn;
    }

    // 还连接
    public function put(RpcConnection $conn): void
    {
        // 连接坏了，不还池子，直接丢掉
        if (!$conn->healthy || !$conn->isConnected()) {
            $conn->close();
            $this->currentSize--;
            echo "坏连接已丢弃，当前连接数：{$this->currentSize}\n";

            // 如果低于最小值，补一个
            if ($this->currentSize < $this->minSize) {
                $newConn = $this->makeConnection();
                if ($newConn) $this->channel->push($newConn);
            }
            return;
        }

        $this->channel->push($conn);
    }

    // 用法语法糖：自动取还，不用手动put
    public function use(callable $fn): mixed
    {
        $conn = $this->get();
        try {
            return $fn($conn);
        } catch (\Throwable $e) {
            $conn->healthy = false;
            throw $e;
        } finally {
            $this->put($conn);
        }
    }

    // 新建连接
    private function makeConnection(): ?RpcConnection
    {
        try {
            $conn = new RpcConnection($this->host, $this->port);
            $this->currentSize++;
            return $conn;
        } catch (\Throwable $e) {
            echo "建连接失败：{$e->getMessage()}\n";
            return null;
        }
    }

    // 定时回收空闲连接（低于minSize不回收）
    private function startIdleChecker(): void
    {
        Swoole\Timer::tick(30000, function () {
            $checked = 0;
            $total   = $this->channel->length();

            // 逐个检查池子里的连接
            while ($checked < $total && $this->channel->length() > 0) {
                $conn = $this->channel->pop(0.001);
                if (!$conn) break;

                $idleSeconds = microtime(true) - $conn->lastUsed;
                $isIdle      = $idleSeconds > $this->idleTime;
                $canShrink   = $this->currentSize > $this->minSize;

                // 空闲太久 且 超过最小连接数 → 回收
                if ($isIdle && $canShrink) {
                    $conn->close();
                    $this->currentSize--;
                    echo "回收空闲连接，当前连接数：{$this->currentSize}\n";
                } else {
                    // 还正常，放回去
                    $this->channel->push($conn);
                }
                $checked++;
            }
        });
    }

    public function stats(): array
    {
        return [
            'current_size' => $this->currentSize,
            'idle'         => $this->channel->length(),
            'busy'         => $this->currentSize - $this->channel->length(),
        ];
    }
}

<?php
// server.php
require __DIR__ . '/Pool.php';

$server = new Swoole\Http\Server('0.0.0.0', 9515);
$server->set(['worker_num' => 4]);

$pool = null;

$server->on('workerStart', function () use (&$pool) {
    // 每个worker独立一个连接池
    $pool = new ConnectionPool(
        host:     'rpc服务IP',
        port:     9516,
        minSize:  5,
        maxSize:  50,
        idleTime: 60,
        waitTime: 3.0,
    );
});

$server->on('request', function ($req, $resp) use (&$pool) {
    try {
        // 用法：自动取连接、自动还回去
        $result = $pool->use(function (RpcConnection $conn) use ($req) {
            return $conn->send(json_encode([
                'method' => 'getUser',
                'id'     => $req->get['id'] ?? 1,
            ]));
        });

        $resp->end(json_encode([
            'result' => $result,
            'pool'   => $pool->stats(),
        ]));

    } catch (\Throwable $e) {
        $resp->status(500);
        $resp->end(json_encode(['error' => $e->getMessage()]));
    }
});

$server->start();

扩缩容示意图

启动时：
Channel: [连接1][连接2][连接3][连接4][连接5]   currentSize=5

并发上来了（10个请求同时来）：
Channel: []    currentSize=10（自动扩到10）
请求1取走连接1
请求2取走连接2
...
请求6来了→池子空了→新建连接6→取走     ← 自动扩容
请求11来了→currentSize==maxSize→等待   ← 到上限了等

请求处理完，连接还回来：
Channel: [连接1][连接2]...[连接10]

闲置60秒后（定时器检查）：
Channel: [连接1][连接2][连接3][连接4][连接5]   ← 缩回minSize

Channel为什么天然适合做连接池

// 普通队列（需要自己加锁）：
$mutex->lock();
$conn = array_pop($pool);
$mutex->unlock();

// Swoole Channel（协程安全，不用加锁）：
$conn = $channel->pop();  // 空了自动挂起等待
$channel->push($conn);    // 有数据自动唤醒等待的协程

核心三句话

1. Channel做队列  → 天然协程安全，pop空了自动等，push了自动唤醒
2. use()语法糖    → 自动取还连接，异常也能还，不会泄露连接
3. 定时缩容       → 低于minSize不动，超过的空闲连接定时清理