从零写一个连接池:为什么不用 std::queue

代码:src/pool/connection_pool.{h,cc} (350 行) 场景:rpc_client 每个请求都要 TCP 连接,per-call connect 性能 7,800 QPS,加连接池 28,000 QPS

RPC 客户端最朴素的写法:connect() → send() → recv() → close()。性能 7,800 QPS(p99 4.2ms)。

上连接池,瞬间 28,000 QPS(p99 2.1ms)。3.5x 加速。

听起来就是 std::queue<TcpConnection*> + mutex 的事。真写起来,光"连接怎么回收"就够喝一壶。

一、为什么不用 `std::queue`

初版我就用的 std::queue:

1
2
3
4
5
class ConnectionPool {
  std::queue<std::unique_ptr<TcpConnection>> pool_;
  std::mutex mu_;
  std::condition_variable cv_;
};

跑起来有问题:

close 阻塞:~ConnectionPool 析构时,如果有线程在 acquire() 阻塞等连接,会卡死——cv_.wait 等不到 notify。
健康连接没法驱逐:连接挂掉(对端 RST)放回池子,下次拿出来用直接报错。
没有 per-thread 缓存:高并发下,100 个线程抢 1 把锁 = 锁竞争。
没有空闲淘汰:池子里 50 个连接空闲,占着 fd 不放。

所以,正经的连接池至少要 7 个细节。

二、连接池的 7 个设计点

2.1 连接健康度:怎么判定"挂了"

问题:tcp_connection_->send(data) 失败时,怎么知道是对端 RST / 网络断 / 业务拒绝?

最朴素:返回的 errno

EPIPE / ECONNRESET → 对端关了,丢
ETIMEDOUT → 慢,丢
EAGAIN → 暂时不可写,等

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
bool TcpConnection::send(Buffer& buf) {
  ssize_t n = ::send(fd_, buf.peek(), buf.readableBytes(), MSG_NOSIGNAL);
  if (n < 0) {
    if (errno == EAGAIN) return false;  // 慢,不算挂
    if (errno == EPIPE || errno == ECONNRESET) {
      healthy_ = false;  // 挂了
      return false;
    }
  }
  return true;
}

2.2 超时回收:`acquire(timeout)`

问题:acquire() 应该带超时——客户端不可能永远等。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
TcpConnection* acquire(int timeout_ms) {
  std::unique_lock<std::mutex> lk(mu_);
  if (!cv_.wait_for(lk, std::chrono::milliseconds(timeout_ms), 
                    [this] { return !pool_.empty() || closed_; })) {
    return nullptr;  // timeout
  }
  if (closed_) return nullptr;
  auto conn = std::move(pool_.front());
  pool_.pop();
  return conn.release();
}

关键:wait_for 第三个参数是 lambda——wait 到条件为 true 才返回,不会"等够 1 秒再看,可能已经有人放了"。

2.3 空闲淘汰:LRU?

问题:池子最大 100,但实际负载只用到 10,90 个空着浪费 fd。

最简方案:最大空闲时间。acquire() 时清理超过 60s 没用的:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
TcpConnection* acquire(int timeout_ms) {
  std::unique_lock<std::mutex> lk(mu_);
  auto now = Clock::now();
  while (!pool_.empty()) {
    auto& conn = pool_.front();
    if (now - conn->last_used() > 60s) {
      // 过期,关掉
      conn.reset();
      pool_.pop();
      continue;
    }
    break;
  }
  // ... 后续 acquire
}

没上 LRU——LRU 需要双向链表,代码复杂度上去,实际收益不大。简单 LIFO 淘汰足够。

2.4 per-thread 缓存:减少锁竞争

问题:100 线程抢 1 把锁,acquire 成为瓶颈。

方案:per-thread 缓存 + 共享池。每线程先看自己缓存,没有才去抢共享池。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
thread_local std::vector<std::unique_ptr<TcpConnection>> local_cache_;

// acquire 流程
TcpConnection* acquire(int timeout_ms) {
  // 1. 看本线程缓存
  if (!local_cache_.empty()) {
    auto c = local_cache_.back().release();
    local_cache_.pop_back();
    return c;
  }
  // 2. 抢共享池
  return shared_pool_.acquire(timeout_ms);
}

实测:128 线程下,加 per-thread 缓存 QPS 再涨 30%。

但有泄漏风险——线程退出时,thread_local 析构,连接直接 close(没放回池子)。可以接受(线程是常驻的 EventLoop),但要记录per-thread 缓存的最大数。

2.5 异步 acquire:怎么 wake up

问题:同步 acquire(timeout_ms) 简单,但调用方写起来啰嗦:

1
2
auto conn = pool->acquire(100);
if (!conn) { /* retry? 报错? */ }

异步版:返回 std::future<TcpConnection*> 或 callback:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
std::future<TcpConnection*> acquire_async(int timeout_ms) {
  auto p = std::make_shared<std::promise<TcpConnection*>>();
  auto fut = p->get_future();
  
  // 入队等待
  {
    std::lock_guard<std::mutex> lk(mu_);
    if (!pool_.empty()) {
      p->set_value(pool_.front().release());
      pool_.pop();
      return fut;
    }
    waiters_.push_back(p);
  }
  
  // 等待超时线程负责 set exception
  schedule_after(timeout_ms, [p] { 
    p->set_value(nullptr); 
  });
  
  return fut;
}

我最终没上异步——同步 + 超时已经够用,代码复杂度低。生产里如果要 await,可以让用户自己包 std::async。

2.6 close 时的等待者

问题:~ConnectionPool 析构时,有 50 个线程在 acquire 里等。怎么不卡死?

关键:closed_ 标志 + cv_.notify_all:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
~ConnectionPool() {
  {
    std::lock_guard<std::mutex> lk(mu_);
    closed_ = true;
  }
  cv_.notify_all();  // 唤醒所有 waiter
  
  // 等所有 waiter 退出
  // (实际靠 use_count 或 wait_group,这里简化)
  
  // 关闭所有连接
  for (auto& c : pool_) c->close();
}

acquire 里检测 closed_,返回 nullptr,调用方检查并退出。

2.7 连接泄漏怎么排查

问题:写着写着,fd 用了 10000 个,没释放。

诊断工具:

lsof -p <pid> — 看进程 fd
/proc/<pid>/fdinfo/ — 看具体 fd 状态
连接池自带 metric:pool_size() / acquire_count / release_count / broken_count

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
// pool 内部
struct PoolStats {
  std::atomic<uint64_t> acquire_count{0};
  std::atomic<uint64_t> release_count{0};
  std::atomic<uint64_t> broken_count{0};
  std::atomic<uint64_t> timeout_count{0};
  std::atomic<int> live_size{0};   // 当前在外的连接数
};

PoolStats stats_;

// acquire 时
void on_acquire() {
  stats_.acquire_count++;
  stats_.live_size++;
}

// release 时
void on_release(bool healthy) {
  stats_.release_count++;
  stats_.live_size--;
  if (!healthy) stats_.broken_count++;
}

live_size 不下降 = 泄漏。broken_count 暴涨 = 对端在频繁 RST(可能是负载均衡器健康检查问题)。

三、完整代码(简化)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
// connection_pool.h
class ConnectionPool {
 public:
  explicit ConnectionPool(std::string host, uint16_t port,
                          size_t max_size = 100, 
                          std::chrono::seconds idle_timeout = std::chrono::seconds(60));
  ~ConnectionPool();

  // 同步 acquire,带超时
  class Handle;  // RAII 包装,析构自动 release
  std::optional<std::unique_ptr<Handle>> acquire(std::chrono::milliseconds timeout);

  // metric
  struct Stats { /* ... */ };
  Stats stats() const;

 private:
  std::unique_ptr<Handle> do_acquire(std::chrono::milliseconds timeout);
  void do_release(std::unique_ptr<TcpConnection> conn, bool healthy);

  std::string host_;
  uint16_t port_;
  size_t max_size_;
  std::chrono::seconds idle_timeout_;

  std::mutex mu_;
  std::condition_variable cv_;
  std::deque<std::unique_ptr<TcpConnection>> pool_;
  bool closed_ = false;

  std::atomic<uint64_t> next_id_{0};
  Stats stats_;
};

Handle 用 RAII 包装:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
class Handle {
 public:
  Handle(ConnectionPool* pool, std::unique_ptr<TcpConnection> conn)
    : pool_(pool), conn_(std::move(conn)) {}
  ~Handle() {
    if (conn_) pool_->release(std::move(conn_), healthy_);
  }
  TcpConnection* get() { return conn_.get(); }
  void mark_broken() { healthy_ = false; }
 private:
  ConnectionPool* pool_;
  std::unique_ptr<TcpConnection> conn_;
  bool healthy_ = true;
};

用法:

1
2
3
4
auto handle = pool.acquire(100ms);
if (!handle) { /* timeout */ return error; }
(*handle)->send(buf);  // 用连接
// ~Handle 自动 release 到池子

四、性能对比

方案	QPS	p99	说明
每请求 connect/close	7,800	4.2ms	握手 + close syscalls
共享池 + mutex	14,500	1.8ms	锁竞争
共享池 + per-thread 缓存	28,000	0.8ms	几乎无锁
共享池 + per-thread 缓存(8 连接 / 8 线程)	48,765	0.38ms	连接数与线程数对齐

3.5x → 6.2x,全程无新库,纯 std + epoll。

口径说明:7,800 / 14,500 / 28,000 是开发机上逐版迭代的实测(早期版本未留压测程序);仓库 tests/test_benchmark.cc 当前可复现的口径为 direct 长连接 30,502 QPS → 池化 8 连接 48,765 QPS(+60%),p99 382μs。

五、最容易踩的 3 个坑

5.1 忘记 close 失效的连接

1
2
3
4
5
6
7
void release(std::unique_ptr<TcpConnection> conn, bool healthy) {
  if (!healthy) {
    conn->close();  // 必须关!
  } else {
    pool_.push_back(std::move(conn));
  }
}

不关的话,fd 泄漏——broken 连接放回池子,下次拿出来 send 立刻又 broken,永远不通。

5.2 thread_local 缓存上限

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
thread_local std::vector<std::unique_ptr<TcpConnection>> local_cache_;

auto* acquire() {
  if (!local_cache_.empty()) {
    auto c = local_cache_.back().release();
    local_cache_.pop_back();
    return c;
  }
  // ...
}

void release(TcpConnection* c) {
  if (local_cache_.size() < 16) {  // 上限!
    local_cache_.emplace_back(c);
  }
}

不上限,thread_local 缓存可能 10000+。上限 + 共享池 fallback 是必须的。

5.3 close 时还在阻塞 send

TcpConnection::send 是阻塞的(在 EventLoop 线程里跑)。连接 close 时,fd 关闭,但 EventLoop 线程可能正在 send。

解决:send 里检查 closed_ flag,关了就立刻返回:

1
2
3
4
bool TcpConnection::send(Buffer& buf) {
  if (closed_) return false;
  // ...
}

或者用 eventfd 唤醒 EventLoop,强制 quit loop。

六、连接池 vs 共享 fd 池:另一种设计

我的池子每个连接一个 fd。也可以"多个请求共享 1 个连接 + pipeline"——HTTP/1.1 pipelining、Redis pipelining 都是这种。

对比:

方案	优点	缺点
每连接一连接	简单、独立	高并发时 fd 多
Pipeline 共享	fd 少,吞吐高	请求必须能异步,编码复杂

我选前者——RPC 调用需要响应顺序与请求顺序一致(没设计成 pipelined),独立连接最简单。生产里如果请求能异步,选 pipeline——单连接能跑到 100K+ QPS。

相关阅读: