Erlang 数据库集成与连接池实战

%% rebar.config
{deps, [{epgsql, "4.7.1"}, {poolboy, "1.5.2"}]}.

-module(pg_demo).
-export([connect/0, query/2]).

connect() ->
    {ok, Conn} = epgsql:connect(#{
        host => "10.0.1.5",
        port => 5432,
        username => "app",
        password => "secret",
        database => "prod",
        timeout => 5000,
        ssl => true
    }),
    Conn.

%% 参数化查询：$1 / $2 占位，自动转义，杜绝注入
find_user(Conn, UserId) ->
    case epgsql:equery(Conn, "SELECT id, name, email FROM users WHERE id = $1",
                       [UserId]) of
        {ok, _Cols, [{Id, Name, Email}]} ->
            {ok, #{id => Id, name => Name, email => Email}};
        {ok, _Cols, []} ->
            {error, not_found};
        {error, Reason} ->
            {error, Reason}
    end.

{ok, Pid} = mysql:start_link([
    {host, "10.0.1.6"},
    {port, 3306},
    {user, "app"},
    {password, "secret"},
    {database, "prod"},
    {query_timeout, 5000},
    {connect_timeout, 5000}
]).

%% 参数化查询：? 占位
mysql:query(Pid, "SELECT id, name FROM users WHERE id = ?", [UserId]).

%% PostgreSQL 的 numeric 默认返回 binary，需显式转换
{ok, _, [{<<"123.45">>}]} = epgsql:equery(Conn, "SELECT 123.45::numeric"),

%% 解决方案：查询时转换，或在驱动层配置
epgsql:equery(Conn, "SELECT amount::float8 FROM orders WHERE id = $1", [Id]).

%% timestamp 返回 {{Y,M,D},{H,Mi,S}} 元组
{ok, _, [{{2026,10,1},{12,0,0}}]} = epgsql:equery(Conn, "SELECT now()::timestamp").

%% 反模式：每次请求新建连接
handle_request(Req) ->
    {ok, Conn} = epgsql:connect(Opts),      % 昂贵！
    Result = epgsql:equery(Conn, Sql, Params),
    epgsql:close(Conn),
    Result.

%% 正解：从池中借出，用完归还
handle_request(Req) ->
    poolboy:transaction(db_pool, fun(Worker) ->
        gen_server:call(Worker, {query, Sql, Params})
    end).

%% 池规格（放入应用监督树）
PoolArgs = [
    {name, {local, db_pool}},
    {worker_module, pg_worker},
    {size, 16},                % 常驻连接数
    {max_overflow, 8}          % 峰值可临时新建
],
ChildSpec = poolboy:child_spec(db_pool, PoolArgs, PgConnectArgs),
{ok, {SupFlags, [ChildSpec]}}.

%% pg_worker.erl：每 worker 持有一条连接
-module(pg_worker).
-behaviour(gen_server).
-behaviour(poolboy_worker).
-export([init/1, handle_call/3, handle_cast/2, terminate/2]).

init(ConnectArgs) ->
    {ok, Conn} = epgsql:connect(ConnectArgs),
    {ok, #{conn => Conn}}.

handle_call({query, Sql, Params}, _From, #{conn := Conn} = State) ->
    {reply, epgsql:equery(Conn, Sql, Params), State};
handle_call(_Req, _From, State) ->
    {reply, {error, unknown}, State}.

handle_cast(_Msg, State) -> {noreply, State}.

terminate(_Reason, #{conn := Conn}) -> catch epgsql:close(Conn), ok.

%% 借出时设置等待超时，避免请求无限挂起
case poolboy:checkout(db_pool, false, 1000) of
    Worker when is_pid(Worker) ->
        try gen_server:call(Worker, {query, Sql, Params})
        after poolboy:checkin(db_pool, Worker)
        end;
    full ->
        {error, pool_exhausted}
end.

get_user_cached(UserId) ->
    case ets:lookup(user_cache, UserId) of
        [{_, User}] -> {ok, User};
        [] ->
            {ok, User} = poolboy:transaction(db_pool, fun(W) ->
                gen_server:call(W, {query_user, UserId})
            end),
            ets:insert(user_cache, {UserId, User}),
            {ok, User}
    end.

%% 事务必须绑定在同一条连接上，因此整个事务在 worker 进程内完成
transfer(From, To, Amount) ->
    poolboy:transaction(db_pool, fun(Worker) ->
        gen_server:call(Worker, {transaction, fun(Conn) ->
            {ok, _, _} = epgsql:equery(Conn,
                "UPDATE accounts SET balance = balance - $1 WHERE id = $2",
                [Amount, From]),
            {ok, _, _} = epgsql:equery(Conn,
                "UPDATE accounts SET balance = balance + $1 WHERE id = $2",
                [Amount, To]),
            ok
        end}, 10000)
    end).

%% worker 侧：用 sqerl 风格的事务包装（伪代码示意语义）
handle_call({transaction, Fun}, _From, #{conn := Conn} = State) ->
    Result = epgsql:with_transaction(Conn, fun(C) -> Fun(C) end, [{reraise, true}]),
    {reply, Result, State}.

%% PostgreSQL：服务端强制语句超时，防止长查询拖垮池
%% SET statement_timeout = '5s';
%% SET idle_in_transaction_session_timeout = '10s';

%% 1. 解析（返回语句名与参数类型）
{ok, Statement} = epgsql:parse(Conn,
    "INSERT INTO events (type, payload) VALUES ($1, $2)", []),

%% 2. 多次执行（复用已解析的计划）
epgsql:prepared_query(Conn, Statement, [<<"order.created">>, Payload1]),
epgsql:prepared_query(Conn, Statement, [<<"order.paid">>, Payload2]),

%% 3. 批量插入（COPY 比逐条 insert 快一个数量级）
epgsql:copy_from(Conn, "COPY events (type, payload) FROM STDIN",
                 [{<<"order.created">>, <<"{}">>} | MoreRows]).

%% 批量插入：拼一条多值 INSERT，比 N 次单条快得多
batch_insert(Conn, Rows) ->
    Placeholders = string:join(
        [io_lib:format("($~p, $~p)", [I * 2 + 1, I * 2 + 2])
         || I <- lists:seq(0, length(Rows) - 1)], ", "),
    Sql = "INSERT INTO events (type, payload) VALUES " ++ Placeholders,
    Params = lists:flatten(Rows),
    epgsql:equery(Conn, Sql, Params).

%% 大结果集分页，避免一次性加载
page(Conn, Limit, Offset) ->
    epgsql:equery(Conn,
        "SELECT id, name FROM users ORDER BY id LIMIT $1 OFFSET $2",
        [Limit, Offset]).

-- migrations/20261001_create_orders.sql
CREATE TABLE IF NOT EXISTS orders (
    id          BIGSERIAL PRIMARY KEY,
    user_id     BIGINT NOT NULL REFERENCES users(id),
    amount      NUMERIC(12,2) NOT NULL,
    status      TEXT NOT NULL DEFAULT 'pending',
    created_at  TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_orders_user_created
    ON orders (user_id, created_at DESC);

# dbmate 常用命令
dbmate new create_orders          # 生成迁移文件
dbmate up                         # 执行未应用的迁移
dbmate status                     # 查看状态
dbmate rollback                   # 回滚最近一次

%% 用 EXPLAIN 验证索引是否生效
{ok, _, Rows} = epgsql:equery(Conn,
    "EXPLAIN (ANALYZE, BUFFERS) SELECT * FROM orders WHERE user_id = $1", [UserId]),
%% 关注输出中的 Index Scan / Seq Scan / Rows Removed by Filter

%% 1. 连接建立超时
{ok, Conn} = epgsql:connect(#{timeout => 5000, ...}),

%% 2. 语句执行超时（客户端侧）
epgsql:equery(Conn, Sql, Params, 5000),

%% 3. 服务端侧超时（最可靠，连接归还后自动清理）
%% SET statement_timeout = '5s';

%% 连接断开时 worker 崩溃，由 poolboy 监督树自动重建
%% 关键：确保 terminate 里关闭旧连接，init 里新建连接
init(ConnectArgs) ->
    case epgsql:connect(ConnectArgs) of
        {ok, Conn} -> {ok, #{conn => Conn}};
        {error, Reason} -> {stop, {connect_failed, Reason}}
    end.

terminate(_Reason, #{conn := Conn}) ->
    catch epgsql:close(Conn),      % 幂等关闭，忽略已断开错误
    ok.

%% 数据库不可用时的降级：池满或超时都快速失败，不无限等待
get_user(UserId) ->
    case poolboy:checkout(db_pool, false, 500) of
        full -> {error, service_unavailable};
        Worker ->
            try gen_server:call(Worker, {query_user, UserId}, 3000)
            catch exit:{timeout, _} -> {error, timeout}
            after poolboy:checkin(db_pool, Worker)
            end
    end.