PHP小码农

Appearance

Node.js 连接 PostgreSQL ​

Node.js 生态连接 PostgreSQL 的两大原生驱动：pg（node-postgres，事实标准，API 经典、生态最广）与 postgres（postgres.js，porsager 出品，模板字符串 SQL 风格，DX 极佳）。在此之上，Prisma、Knex、TypeORM 等 ORM/查询构建器覆盖了不同抽象层级的需求。本页以 pg 为主线——连接池、事务、游标、类型解析、LISTEN/NOTIFY、批量插入与 COPY——再用 postgres.js 与 ORM 概览补全选型视野。

Node.js 的运行模型（单线程事件循环）与 PostgreSQL 的连接模型（每连接一个服务端进程）组合出本页最重要的两条纪律：查询绝不阻塞事件循环（大结果集必须流式或分块取回），连接必须池化（每个并发请求若各自建连接，PostgreSQL 后端进程数会瞬间失控）。这两条纪律的落地手段——pg 的 Pool、pg-cursor 游标、pg-copy-streams 流式 COPY——都会在正文中一一展开。

代码示例以 pg v8 与 Node.js 20 LTS 为基准，postgres.js 与 Prisma 部分标注版本背景；API 细节以各自官方文档为准。

前置知识

建议先阅读：Go 连接 PostgreSQL

基础概念 ​

方案定位风格
pg（node-postgres）事实标准驱动经典回调/Promise API，Client/Pool
postgres（postgres.js）现代 DX 驱动模板字符串 SQL，内建池与事务助手
PrismaORM + 迁移工具链schema 声明式建模
Knex查询构建器 + 迁移链式 SQL 构建
TypeORM传统 ORM装饰器实体
text
  ┌───────────────┐
  │ Express/Fastify│──► pg Pool ──────────────► PostgreSQL
  │  (事件循环)     │     (复用 N 条连接)
  └───────────────┘
  ┌───────────────┐
  │ 业务服务        │──► Prisma / Knex / TypeORM ──► pg 或自有驱动 ──► PostgreSQL
  └───────────────┘

三个语言层面的关键认知：

  1. $1 占位符：pg 与 postgres.js 的参数都走服务端绑定（扩展查询协议），值与语句分离传输，注入防护落在协议层；
  2. 事件循环模型：一条阻塞 CPU 的行转换逻辑会拖慢所有并发请求——「快取快放、分块处理」是 Node + PostgreSQL 的内存与延迟策略；
  3. 异步处处 await：事务里 await client.query('BEGIN') 与普通查询没有语法差异，错误路径的 ROLLBACK 必须用 try/catch/finally 保证执行。

语法与用法 ​

pg 的池与查询：

javascript
const { Pool } = require('pg');

const pool = new Pool({
  host: 'db1.prod.internal',
  port: 5432,
  database: 'appdb',
  user: 'app_rw',
  password: process.env.DB_PASSWORD,
  max: 20,                        // 池内最大连接数
  idleTimeoutMillis: 30_000,      // 空闲回收
  connectionTimeoutMillis: 3_000, // 借连接超时
  ssl: { rejectUnauthorized: true },
});

const { rows } = await pool.query(
  'SELECT id, name FROM users WHERE status = $1 AND age > $2',
  ['active', 18],
);

事务模式（从池借独占连接）：

javascript
const client = await pool.connect();
try {
  await client.query('BEGIN');
  await client.query(
    'UPDATE accounts SET balance = balance - $1 WHERE id = $2',
    [100, fromId],
  );
  await client.query(
    'UPDATE accounts SET balance = balance + $1 WHERE id = $2',
    [100, toId],
  );
  await client.query('COMMIT');
} catch (e) {
  await client.query('ROLLBACK');
  throw e;
} finally {
  client.release();
}

批量与 COPY：

javascript
// 多值展开插入（中批量）
const values = [];
const chunks = items.map((it, i) => {
  values.push(it.id, it.name);
  return `($${i * 2 + 1}, $${i * 2 + 2})`;
});
await pool.query(`INSERT INTO items (id, name) VALUES ${chunks.join(',')}`, values);

// COPY 流式导入（大批量，pg-copy-streams）
const { from: copyFrom } = require('pg-copy-streams');
const fs = require('fs');
const client = await pool.connect();
try {
  const stream = client.query(
    copyFrom('COPY stg_items FROM STDIN WITH (FORMAT csv, HEADER true)'),
  );
  await new Promise((resolve, reject) => {
    fs.createReadStream('items.csv')
      .pipe(stream)
      .on('finish', resolve)
      .on('error', reject);
  });
} finally {
  client.release();
}

postgres.js 的模板字符串风格：

javascript
const postgres = require('postgres');
const sql = postgres('postgres://app_rw:secret@db1:5432/appdb', { max: 20 });

const users = await sql`SELECT id, name FROM users WHERE status = ${status}`;
await sql.begin(async (tx) => {
  await tx`UPDATE accounts SET balance = balance - ${amount} WHERE id = ${fromId}`;
  await tx`UPDATE accounts SET balance = balance + ${amount} WHERE id = ${toId}`;
});
await sql.end();

详细说明 ​

pg：Client 与 Pool ​

Client 是单连接对象（脚本、工具、临时任务），Pool 是生产标配：

javascript
// 单连接（用完即关）
const { Client } = require('pg');
const client = new Client({ connectionString: process.env.DATABASE_URL });
await client.connect();
// …查询…
await client.end();

// 池（服务常驻）
const pool = new Pool({ connectionString: process.env.DATABASE_URL, max: 20 });
// pool.query 内部自动借还连接，适合无状态的单条查询
const res = await pool.query('SELECT 1');

pool.query() 把「借连接 → 执行 → 归还」收敛成一步，适合单条语句；事务与需要独占连接的操作必须 pool.connect() 显式借出，用完 release() 归还。连接配置既可用独立字段也可用 connectionString（libpq URL），ssl: { rejectUnauthorized: false } 仅用于内网自签证书调试，公网环境必须校验证书。

pg：查询、参数与结果结构 ​

pool.query(text, values) 返回 { rows, rowCount, fields }：

javascript
const res = await pool.query('SELECT id, name FROM users LIMIT 5');
// res.rows: 行数组（每行是对象，键为列名）
// res.rowCount: 受影响/返回行数
// res.fields: 列元信息

// INSERT/UPDATE/DELETE 用 rowCount
const del = await pool.query('DELETE FROM sessions WHERE expired_at < now()');
console.log(`removed ${del.rowCount} rows`);

值不能参数化的位置照旧要白名单：表名、列名、排序方向由代码白名单映射，LIMIT/OFFSET 整数化：

javascript
const SORTS = { created: 'created_at', amount: 'amount' };
const col = SORTS[req.query.sort] ?? 'created_at';
const limit = Math.min(Number(req.query.limit) || 20, 100);
const sql = `SELECT * FROM orders ORDER BY ${col} DESC LIMIT ${limit}`;

pg：事务的标准模式 ​

事务必须独占连接，pg 的经典模板（见语法节）有三个不可省略的部分：

  1. BEGIN 之后的每条语句都用同一个 client——用 pool.query 会借到不同连接，事务形同虚设（新手第一坑）；
  2. catch 里 ROLLBACK：任何语句失败都进入 catch，先回滚再决定抛出或重试；
  3. finally 里 release()：忘释放的连接不会归还池，泄漏积累到 max 后全部请求排队超时。

进阶：SAVEPOINT 与嵌套事务在 pg 里就是 SQL（SAVEPOINT sp1 / ROLLBACK TO sp1）；Pool 的 on('error') 事件处理空闲连接意外断开（数据库重启等），避免未处理异常崩溃进程。

pg：pg-cursor 游标与大结果集 ​

pool.query 会把全部结果缓冲在内存；几十万行以上的查询用 pg-cursor 分块取回：

javascript
const Cursor = require('pg-cursor');
const client = await pool.connect();
try {
  const cursor = client.query(new Cursor('SELECT * FROM big_table'));
  for (;;) {
    const rows = await new Promise((resolve, reject) =>
      cursor.read(10_000, (err, r) => (err ? reject(err) : resolve(r))),
    );
    if (rows.length === 0) break;
    process(rows);            // 每批 1 万行，内存平稳
  }
  await new Promise((resolve) => cursor.close(resolve));
} finally {
  client.release();
}

游标在服务端逐块推进（声明为 WITHOUT HOLD 的服务端游标），连接在游标生命周期内独占——处理完尽快 close + release。

pg：类型解析 pg.types.setTypeParser ​

pg 默认把一些类型解析得「出乎意料」：NUMERIC 返回字符串（保护精度，合理）、INT8/bigint 返回字符串（防 JS 精度溢出）、DATE/TIMESTAMP 按 JS Date（时区按本地/UTC 的组合规则）。用 pg.types.setTypeParser 按需改写：

javascript
const pg = require('pg');
const types = pg.types;

// bigint 超出 Number.MAX_SAFE_INTEGER 的风险自担；确定安全时可解析为 number
types.setTypeParser(types.builtins.INT8, (v) => parseInt(v, 10));
// NUMERIC 保持字符串（默认），配合 decimal.js 计算；也可 parseFloat（精度风险自负）
// types.setTypeParser(types.builtins.NUMERIC, parseFloat);

types.builtins 以 OID 常量提供类型索引（INT8、NUMERIC、TIMESTAMP 等）。改写是进程级全局的，务必在应用初始化时统一设置并写进规范，避免各模块行为不一致。

pg：命名预处理与语句复用 ​

client.query 除 text + values 外还支持命名预处理语句：同一 name 的语句在该连接上只解析一次，后续调用直接复用服务端计划：

javascript
const client = await pool.connect();
try {
  // 首次：解析并缓存为命名语句
  await client.query({
    name: 'get-user-by-email',
    text: 'SELECT id, name FROM users WHERE email = $1',
    values: ['a@example.com'],
  });
  // 后续：复用计划，仅传参数
  await client.query({
    name: 'get-user-by-email',
    values: ['b@example.com'],
  });
} finally {
  client.release();
}

命名语句按连接缓存——借到不同连接就要重新解析，这与 PgBouncer transaction 模式的冲突（见《PHP 连接 PostgreSQL》同名小节）同源。工程口径：依赖 pool.query 的匿名语句即可，pg 的自动计划管理覆盖多数场景；确有热路径需要固定计划时，为该查询申请专用连接，而不是把命名语句撒进池里。

pg：LISTEN/NOTIFY ​

javascript
// 独占一个专用连接做订阅
const listener = new Client({ connectionString: process.env.DATABASE_URL });
await listener.connect();
listener.on('notification', (msg) => {
  console.log(`channel=${msg.channel} payload=${msg.payload}`);
});
await listener.query('LISTEN order_events');

// 通知方（任意连接）
await pool.query("NOTIFY order_events, '{\"order_id\":42}'");

语义与其他语言一致：独占连接、事务提交后投递、无持久化与重放。Node 侧的注意点是专用连接独立于池管理（不进 Pool，或从 Pool 长期借出不归还），并处理连接断开重连。

批量插入与 COPY 的分档 ​

量级手段说明
≤ 数百行逐条 await 或 Promise.all（受池上限约束）简单直接
数百~数万行多值展开 INSERT（占位符拼接）单条语句、服务端一次解析
数万行以上pg-copy-streams 流式 COPY内存平稳、速度快一个量级

多值展开要注意占位符编号与参数数组的一致性（见语法节模板）；pg-copy-streams 把客户端文件流/生成流直接接进 COPY 通道，copyFrom（导入）与 copyTo（导出）都是 Node Stream——与文件系统流、HTTP 流天然组合。

postgres.js：模板字符串 SQL ​

postgres.js 的核心理念是「模板字符串即参数化」：标签函数自动把 ${} 插值转成 $n 绑定参数，写法上像拼接、语义上是参数化，兼顾 DX 与安全：

javascript
const sql = postgres(process.env.DATABASE_URL, {
  max: 20,
  idle_timeout: 30,
  connect_timeout: 5,
  ssl: 'require',
});

// 基础查询
const users = await sql`SELECT * FROM users WHERE status = ${status}`;

// 单行/标量
const [user] = await sql`SELECT * FROM users WHERE id = ${id}`;
const [{ count }] = await sql`SELECT count(*)::int AS count FROM orders`;

// 批量插入：数组自动展开
await sql`INSERT INTO items ${sql(items)}`;

// 事务
await sql.begin(async (tx) => {
  await tx`UPDATE accounts SET balance = balance - ${amount} WHERE id = ${from}`;
  await tx`UPDATE accounts SET balance = balance + ${amount} WHERE id = ${to}`;
});

// 确有需要时的动态 SQL：sql.unsafe（参数仍可绑定）
const rows = await sql.unsafe('SELECT * FROM t WHERE id = $1', [id]);

内建连接池、sql.begin 事务助手与数组/对象值助手让样板代码极少；它是 pg 之外值得认真评估的选项，API 细节以官方仓库为准。

Prisma / Knex / TypeORM 概述 ​

三个常用抽象层的定位差异：

prisma
// Prisma：schema 声明式建模 + 类型安全的生成客户端
datasource db {
  provider = "postgresql"
  url      = env("DATABASE_URL")
}

model User {
  id        Int      @id @default(autoincrement())
  email     String   @unique
  createdAt DateTime @default(now()) @map("created_at")
  @@map("users")
}
javascript
// Prisma 查询（生成的类型安全客户端）
const users = await prisma.user.findMany({
  where: { status: 'active' },
  orderBy: { createdAt: 'desc' },
  take: 20,
});
  • Prisma：schema 即文档，生成 TS 类型；迁移工具链 prisma migrate 与数据库变更管理打通，适合 TS 团队快速建模；
  • Knex：链式查询构建器（knex('orders').where(...).limit(20)）+ 迁移脚本，抽象克制、贴近 SQL，PostgreSQL 的 JSONB/数组/ON CONFLICT 都有 API；
  • TypeORM：传统 ORM（装饰器实体、Repository、QueryBuilder），适合 JPA 背景团队；注意其连接池配置同样要按「实例数 × 池大小 ≤ max_connections」做总账。

ORM 层的性能规则与裸驱动一致：N+1 查询用 include/join 解决、批量写入走原生 SQL 逃生舱（prisma.$executeRaw、knex.raw）、COPY 这类通道永远在驱动层。

连接池大小与 ssl ​

池大小是本页最重要的容量参数。PostgreSQL 官方对「理想连接数」的经典启发式：(2 × CPU 核数) + 磁盘主轴数——连接过多带来的上下文切换会抵消并发收益。Node 侧的推导链：应用实例数 × pool.max ≤ max_connections 安全余量，多实例/多服务共享数据库时把总账交给 PgBouncer 收敛，应用池相应收小。ssl 参数按环境分级：

javascript
// 内网自签（调试）
ssl: { rejectUnauthorized: false }
// 生产（校验证书）
ssl: { rejectUnauthorized: true, ca: fs.readFileSync('/etc/ssl/pg-ca.crt').toString() }

错误码 err.code = SQLSTATE ​

pg 的错误对象直接携带 SQLSTATE 与丰富的上下文字段：

javascript
try {
  await pool.query('INSERT INTO users (email) VALUES ($1)', [email]);
} catch (err) {
  switch (err.code) {
    case '23505': // unique_violation
      return res.status(409).json({ error: 'email exists', constraint: err.constraint });
    case '23503': // foreign_key_violation
      return res.status(400).json({ error: 'reference missing' });
    case '40001': case '40P01':   // serialization_failure / deadlock_detected
      return retry();
    default:
      throw err;
  }
}

err.constraint（约束名）、err.detail、err.table 让错误可以直接映射到业务语义，比正则匹配错误文本可靠得多。

错误码分支的高频码速查 ​

err.code 最常分支处理的八个码：23505（唯一冲突，转业务 4xx）、23503（外键冲突，校验引用数据）、23514（检查约束，前置校验）、40001（序列化失败，退避重试）、40P01（死锁，退避重试）、57014（查询取消，区分超时来源）、53300（连接超限，退避 + 降级）、08006（连接故障，重建连接）。前两个转业务、中间两个可重试、其余按基础设施告警处理——语义与《PHP 连接 PostgreSQL》的 SQLSTATE 速查表完全一致，跨语言团队共用同一套字典即可。

实战示例 ​

示例一：Pool 初始化 + 事务服务 ​

javascript
// db.js
const { Pool } = require('pg');

const pool = new Pool({
  connectionString: process.env.DATABASE_URL,
  max: 20,
  idleTimeoutMillis: 30_000,
  connectionTimeoutMillis: 3_000,
  ssl: { rejectUnauthorized: true },
});

async function withTransaction(fn) {
  const client = await pool.connect();
  try {
    await client.query('BEGIN');
    const result = await fn(client);
    await client.query('COMMIT');
    return result;
  } catch (e) {
    await client.query('ROLLBACK');
    throw e;
  } finally {
    client.release();
  }
}

module.exports = { pool, withTransaction };
javascript
// 业务调用
const { withTransaction } = require('./db');

await withTransaction(async (client) => {
  const { rows } = await client.query(
    'UPDATE accounts SET balance = balance - $1 WHERE id = $2 AND balance >= $1 RETURNING id',
    [amount, fromId],
  );
  if (rows.length === 0) throw new Error('INSUFFICIENT_BALANCE');
  await client.query(
    'UPDATE accounts SET balance = balance + $1 WHERE id = $2',
    [amount, toId],
  );
});

示例二：pg-copy-streams 导入 CSV ​

javascript
const { from: copyFrom } = require('pg-copy-streams');
const fs = require('fs');
const { pipeline } = require('stream/promises');
const { pool } = require('./db');

async function importCsv(csvPath) {
  const client = await pool.connect();
  try {
    const stream = client.query(
      copyFrom('COPY stg_events FROM STDIN WITH (FORMAT csv, HEADER true)'),
    );
    await pipeline(fs.createReadStream(csvPath), stream);
  } finally {
    client.release();
  }
}

stream/promises.pipeline 保证任一端出错即清理并正确传播错误，避免手写 pipe 事件对。

示例三：postgres.js 完整小型服务 ​

javascript
const postgres = require('postgres');
const sql = postgres(process.env.DATABASE_URL, {
  max: 20, connect_timeout: 5, ssl: 'require',
});

async function getTopOrders(limit = 20) {
  return sql`
    SELECT o.id, o.amount, u.name AS user_name
    FROM orders o JOIN users u ON u.id = o.user_id
    WHERE o.created_at >= now() - interval '7 days'
    ORDER BY o.amount DESC
    LIMIT ${limit}
  `;
}

async function createOrder(payload) {
  return sql.begin(async (tx) => {
    const [order] = await tx`
      INSERT INTO orders (user_id, amount) VALUES (${payload.userId}, ${payload.amount})
      RETURNING id`;
    await tx`DELETE FROM cart_items WHERE user_id = ${payload.userId}`;
    return order;
  });
}

process.on('exit', () => sql.end());

示例四：Prisma schema 与查询 ​

prisma
datasource db {
  provider = "postgresql"
  url      = env("DATABASE_URL")
}

model Order {
  id        Int      @id @default(autoincrement())
  userId    Int
  amount    Decimal  @db.Decimal(12, 2)
  createdAt DateTime @default(now()) @map("created_at")

  @@index([createdAt])
  @@map("orders")
}
javascript
const orders = await prisma.order.findMany({
  where: { createdAt: { gte: new Date('2026-09-01') } },
  orderBy: { amount: 'desc' },
  take: 20,
});

示例五：NOTIFY 订阅与重连 ​

javascript
const { Client } = require('pg');

function startListener() {
  const client = new Client({ connectionString: process.env.DATABASE_URL });
  client.on('error', (e) => { console.error(e); setTimeout(startListener, 3000); });
  client.on('notification', (msg) => handleEvent(msg.channel, msg.payload));
  client.connect().then(() => client.query('LISTEN order_events'));
}

示例六：copyTo 流式导出为压缩文件 ​

javascript
const { to: copyTo } = require('pg-copy-streams');
const fs = require('fs');
const zlib = require('zlib');
const { pipeline } = require('stream/promises');
const { pool } = require('./db');

async function exportEvents() {
  const client = await pool.connect();
  try {
    const source = client.query(
      copyTo('COPY (SELECT * FROM events ORDER BY id) TO STDOUT WITH (FORMAT csv, HEADER true)'),
    );
    await pipeline(source, zlib.createGzip(), fs.createWriteStream('events.csv.gz'));
  } finally {
    client.release();
  }
}

copyTo 返回可读流，与 Node 内建的转换流（gzip、加密、CSV 解析器）直接组管——「SQL 侧排序 → COPY 导出 → 流式压缩落盘」全程不落中间文件，内存占用恒定。导出的 ORDER BY 放进 COPY 的子查询里，让排序发生在数据库侧。

注意事项 ​

事务语句必须走同一个 client。 pool.query('BEGIN') 与后续 pool.query 是不同连接；事务模板的「借出独占连接」不是风格偏好而是正确性要求。

release 必然执行。 try/finally 包住 client.release()；漏 release 的连接泄漏会让池逐渐耗尽，表现为「系统越跑越慢、最终全部 timeout」。

大结果集必须分块。 pool.query 全量缓冲；几十万行用 pg-cursor 或 SQL 层分页/分批，否则一次大查询就能顶爆 Node 进程内存。

类型解析器是进程级全局。 pg.types.setTypeParser 在初始化时统一设置；bigint 默认字符串是有意为之的精度保护，改写前确认业务数值范围。

NUMERIC 返回字符串。 用 JS number 直接算金额会引入浮点误差；配合 decimal.js/bignumber.js 或让 SQL 完成 SUM 等运算。

NOTIFY 不做消息总线。 无持久化、断线即丢、payload 上限 8000 字节；订阅连接要处理 error 事件并重连。

多值展开插入有语句长度与占位符数量上限。 数万行以上改走 COPY；占位符编号手写极易错，封装成工具函数。

池大小做全局连接总账。 实例数 × max ≤ max_connections 余量；多服务共享库时上 PgBouncer，并把应用池调小。

ORM 的逃生舱要预留。 COPY、ON CONFLICT 批量 upsert、窗口函数等场景直接走驱动层原生 SQL，不要在 ORM 里硬绕。

命名预处理语句按连接缓存。 池环境下同一 name 借到不同连接会各自解析一遍；依赖固定计划的热路径用专用连接，其余场景信任驱动的匿名语句管理。

最佳实践 ​

  1. 服务常驻用 Pool、事务显式 connect 借还、finally release。 三句话是 pg 的骨架纪律。
  2. 参数一律 $1 占位符，标识符白名单。 模板字符串驱动的 ${} 是参数化不是拼接——postgres.js 尤其如此。
  3. 事务封装成 withTransaction 工具函数。 团队复用同一套 BEGIN/COMMIT/ROLLBACK/release 模板，消灭手写遗漏。
  4. 批量写入分档：多值展开（中）/ pg-copy-streams（大）。 十万行以上一律 COPY，配合摘索引与 ANALYZE 的通用清单。
  5. pg.types.setTypeParser 统一在启动时配置。 bigint/numeric/timestamp 的解析策略写进代码规范，杜绝模块间不一致。
  6. 错误处理按 err.code（SQLSTATE）分支。 23505 转 409、40001/40P01 触发重试，constraint/detail 字段映射业务语义。
  7. 池参数显式设置并做连接总账。 max、idleTimeoutMillis、connectionTimeoutMillis 三件套 + ssl 分级配置。
  8. LISTEN 用专用连接 + 自动重连。 与池隔离，error 事件驱动重启。
  9. 大导出用 copyTo 流式落盘。 与文件/HTTP 流组合，内存平稳。
  10. ORM 选型按团队语言栈：Prisma（TS 建模）、Knex（SQL 贴近）、TypeORM（传统 ORM）。 无论哪个，复杂批处理都预留原生 SQL 逃生舱。
  11. application_name 与连接可溯源。 URL 里带上服务名，pg_stat_activity 排障一眼定位。
  12. supervise 空闲连接错误。 Pool 的 'error' 事件必须挂处理器，防止数据库重启等事件把进程打崩。

下一节

继续学习：Rust 连接 PostgreSQL

参考链接 ​