数据库连接多路复用

概念概述

数据库连接多路复用（Database Connection Multiplexing）指的是：连接池中间层把大量短命客户端连接复用到少量后端会话上，保护按连接付费的数据库，尤其适用于无服务器架构。

从更严格的定义看，Connection multiplexing is the practice of routing many application connections through a middle tier that maintains a much smaller number of real database sessions, reusing them as clients become idle. 该概念在工程实践中的价值，在于它能帮助团队用统一的语言描述一个具体的质量属性或设计约束，从而在评审与协作中减少歧义。

为什么需要它

It is essential for serverless platforms and any fleet where thousands of short-lived processes would otherwise exhaust the database’s connection limit. 它与 Autovacuum and Dead Tuple Reclamation、LATERAL Joins、Upsert with INSERT … ON CONFLICT 等概念经常一起出现，用于解决相近的工程问题。理解它的适用场景，有助于在多种备选方案之间做出有依据的取舍，而不是凭直觉选择技术方案。

常见误解

• pooling and multiplexing are the same thing
• multiplexing is transparent to every database feature

需要避免的反模式

• multiplexing connections that hold session state across statements
• pointing every client directly at the database and increasing max connections

实践要点

1. A pool keeps sessions open and hands them to callers in turn, while a transaction-capable multiplexer assigns a real session only for the duration of a transaction and returns it afterward.
2. Session-level state such as temporary tables, advisory locks, set variables, or prepared statements travels with the physical session, so a shared session can leak state between unrelated callers.
3. The right pool size is bounded by database CPU and storage, not by application thread count, and pushing beyond that turns latency into queueing rather than throughput.

参考来源

https://www.pgbouncer.org/features.html

代码示例

package store

import (
	"context"
	"database/sql"
)

type User struct {
	ID    int64
	Email string
}

func ListActiveUsers(ctx context.Context, db *sql.DB) ([]User, error) {
	rows, err := db.QueryContext(ctx,
		"SELECT id, email FROM users WHERE active = $1 ORDER BY id", true)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var users []User
	for rows.Next() {
		var user User
		if err := rows.Scan(&user.ID, &user.Email); err != nil {
			return nil, err
		}
		users = append(users, user)
	}
	if err := rows.Err(); err != nil {
		return nil, err
	}
	return users, nil
}

import java.sql.*;
import java.util.ArrayList;
import java.util.List;

public class JdbcDatabaseQuery {

    record User(int id, String name, String email) {}

    public static List<User> findActiveUsers(String url, String user, String pass) {
        String sql = "SELECT id, name, email FROM users WHERE active = ? ORDER BY name";

        List<User> users = new ArrayList<>();
        try (Connection conn = DriverManager.getConnection(url, user, pass);
             PreparedStatement stmt = conn.prepareStatement(sql)) {

            stmt.setBoolean(1, true);
            try (ResultSet rs = stmt.executeQuery()) {
                while (rs.next()) {
                    users.add(new User(
                        rs.getInt("id"),
                        rs.getString("name"),
                        rs.getString("email")
                    ));
                }
            }
        } catch (SQLException e) {
            System.err.println("Database error: " + e.getMessage());
        }
        return users;
    }

    public static void main(String[] args) {
        System.out.println("JDBC query example ready to connect");
    }
}