Node.js 操作 PostgreSQL

本文讲在 Node.js 里如何使用 PostgreSQL：驱动选型、连接池、参数化查询、事务、类型映射、错误处理与生产实践。示例基于 pg（node-postgres），同时给出 ORM 的选型建议。

• 环境：Node.js 20+、PostgreSQL 16、pg 8.x
• 示例默认使用 CommonJS，ESM 写法在文末说明

一、驱动与选型

方案	类型	特点	适用
pg（node-postgres）	驱动	生态最广、支持连接池、COPY、LISTEN	绝大多数场景
postgres（postgres.js）	驱动	API 现代、内置 tag 模板字符串、性能好	偏好简洁 API
Prisma	ORM	类型安全、schema 驱动、迁移工具完善	新项目、重类型安全
TypeORM	ORM	装饰器实体、支持多种数据库	NestJS 项目常见
Drizzle	ORM/Query Builder	轻量、SQL 风格、类型推导好	喜欢贴近 SQL
Knex	Query Builder	灵活、迁移与事务支持	需要动态拼 SQL

选择原则：

• 只要执行 SQL、需要完全控制：用 pg。
• 需要实体映射、迁移、关系加载：用 ORM，但复杂查询仍可回退到原生 SQL。
• 不要在 ORM 里硬写复杂报表 SQL，也不要手写拼接字符串——两者都会带来维护问题。

---

二、安装与连接

npm i pg

2.1 连接字符串

postgres://用户名:密码@主机:端口/数据库?sslmode=disable
postgresql://app:secret@localhost:5432/appdb

2.2 Pool 与 Client

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

const pool = new Pool({
	connectionString: process.env.DATABASE_URL,
	max: 10, // 池内最大连接数
	idleTimeoutMillis: 30000, // 空闲连接回收时间
	connectionTimeoutMillis: 2000, // 获取连接的超时
	statement_timeout: 5000, // 单条 SQL 超时（毫秒）
	application_name: "my-api"
});

// 闲置连接报错（如数据库重启）必须监听，否则进程会崩
pool.on("error", error => {
	console.error("PostgreSQL 空闲连接错误", error.message);
});

module.exports = { pool };

Pool 与 Client 的区别：

	Pool	Client
连接复用	是	否（一个实例一条连接）
并发查询	池内自动分配	需自行排队
事务	用 pool.connect() 取一个 client	直接 client.query('BEGIN')
适用	应用常规查询	事务、LISTEN、长连接

不要用单个 Client 处理所有请求：它是串行的，并发上来会互相阻塞，也无法自动重连。

2.3 SSL

const pool = new Pool({
	connectionString: process.env.DATABASE_URL,
	ssl: { rejectUnauthorized: true } // 生产建议校验证书
});

托管数据库（RDS、云厂商）通常要求 SSL；本地开发一般不启用。连接串里的 sslmode=require 与 ssl 选项同时存在时，以代码配置为准，容易互相覆盖，建议只用一种方式。

---

三、参数化查询

const { pool } = require("./db");

// 命名参数用 $1、$2 占位，值通过数组传入
const result = await pool.query(
	"SELECT id, name, email FROM users WHERE status = $1 AND created_at > $2 ORDER BY created_at DESC LIMIT $3",
	["active", new Date(Date.now() - 7 * 86400000), 20]
);

console.log(result.rows);

规则：

• 永远不要把用户输入拼进 SQL。pg 的参数化会正确转义，拼接则会引入注入。
• 表名、列名、ORDER BY 字段无法用 $1 占位，只能白名单校验后再拼接。
• 一次查询多个语句用分号分隔可以执行，但参数只作用于第一条，不推荐。

白名单示例：

const SORTABLE = new Set(["created_at", "name", "email"]);
const sort = SORTABLE.has(input.sort) ? input.sort : "created_at";
const direction = input.order === "asc" ? "ASC" : "DESC";
const sql = `SELECT * FROM users ORDER BY ${sort} ${direction} LIMIT $1`;
const { rows } = await pool.query(sql, [20]);

3.1 单条查询与取一行

// 只取一行：没有数据时返回 undefined
const { rows } = await pool.query("SELECT * FROM users WHERE id = $1", [id]);
const user = rows[0];

// 用 RETURNING 直接拿到写入结果
const inserted = await pool.query(
	"INSERT INTO users (email, name) VALUES ($1, $2) RETURNING id, created_at",
	[email, name]
);

3.2 预处理语句

// 高频执行的 SQL 可以命名，服务端会缓存执行计划
const statement = {
	name: "get-user-by-id",
	text: "SELECT id, name, email FROM users WHERE id = $1",
	values: [id]
};
const { rows } = await pool.query(statement);

注意：命名语句在同一连接上才有缓存效果，连接池场景下收益有限，且修改 SQL 后必须换名字，否则会命中旧计划。

---

四、结果结构与类型映射

result 的结构：

字段	说明
rows	行数组
rowCount	影响的行数（INSERT/UPDATE/DELETE）
command	执行的命令类型（SELECT、INSERT）
fields	列元信息（名称、类型 OID）

默认类型映射（容易踩坑）：

数据库类型	Node 中的类型	说明
int / smallint	number
bigint / numeric	string	避免精度丢失，默认不转 number
timestamptz / timestamp	Date	时区信息已归一
json / jsonb	object	自动 JSON.parse
boolean	boolean
text / varchar	string
bytea	Buffer
array	Array

处理 numeric 金额：

const { types } = require("pg");

// 1700 是 numeric 的 OID；金额场景建议保留字符串，或用 decimal 库计算
types.setTypeParser(1700, value => (value === null ? null : Number(value)));

// 20 是 bigint 的 OID：仅当确定不会超过 Number.MAX_SAFE_INTEGER 时才转 number
types.setTypeParser(20, value => (value === null ? null : Number(value)));

金额用 Number 会有精度风险，正确做法是保留 string 交给 decimal.js / big.js 计算，或直接在 SQL 里用 numeric 运算。

---

五、事务

const { pool } = require("./db");

async function transfer(fromId, toId, amount) {
	const client = await pool.connect();
	try {
		await client.query("BEGIN");

		const { rowCount } = await client.query(
			"UPDATE accounts SET balance = balance - $1 WHERE id = $2 AND balance >= $1",
			[amount, fromId]
		);
		if (rowCount === 0) {
			throw new Error("余额不足");
		}

		await client.query("UPDATE accounts SET balance = balance + $1 WHERE id = $2", [
			amount,
			toId
		]);

		await client.query("COMMIT");
	} catch (error) {
		await client.query("ROLLBACK");
		throw error;
	} finally {
		client.release(); // 必须归还，否则连接泄漏
	}
}

要点：

1. 事务必须用同一个 client，pool.query 每次可能拿不同连接，事务会失效。
2. finally 里 client.release()，否则池被耗尽后请求全部挂起。
3. release(true) 表示连接已损坏（如网络中断），直接销毁而不是归还。

保存点（局部回滚）：

await client.query("BEGIN");
try {
	await client.query("SAVEPOINT s1");
	await client.query("INSERT INTO logs (...) VALUES (...)");
} catch {
	await client.query("ROLLBACK TO SAVEPOINT s1");
}
await client.query("COMMIT");

隔离级别：

await client.query("BEGIN ISOLATION LEVEL SERIALIZABLE");

串行化冲突会返回错误码 40001，需要应用层重试（见第七节）。

---

六、批量写入

6.1 多值插入

const rows = [
	["a@example.com", "张三"],
	["b@example.com", "李四"]
];

// 动态生成占位符，值仍然参数化
const values = [];
const placeholders = rows.map((row, index) => {
	const base = index * 2;
	values.push(row[0], row[1]);
	return `($${base + 1}, $${base + 2})`;
});

await pool.query(
	`INSERT INTO users (email, name) VALUES ${placeholders.join(", ")}`,
	values
);

单条 SQL 的参数量有上限（PostgreSQL 协议限制约 65535），超大批量要分批，每批 500~1000 行。

6.2 UNNEST（推荐的大批量写法）

const emails = ["a@example.com", "b@example.com"];
const names = ["张三", "李四"];

await pool.query(
	`INSERT INTO users (email, name)
	 SELECT * FROM unnest($1::text[], $2::text[])`,
	[emails, names]
);

数组作为一个参数传入，不受参数个数限制，性能也好。

6.3 COPY（导入大批量数据）

npm i pg-copy-streams

const { from: copyFrom } = require("pg-copy-streams");
const { pipeline } = require("node:stream/promises");
const fs = require("node:fs");

const client = await pool.connect();
try {
	const stream = client.query(copyFrom("COPY users (email, name) FROM STDIN WITH (FORMAT csv, HEADER true)"));
	await pipeline(fs.createReadStream("users.csv"), stream);
} finally {
	client.release();
}

COPY 比 INSERT 快一个数量级，适合初始化与数据迁移。

---

七、错误处理与重试

pg 抛出的错误对象包含 PostgreSQL 的字段：

字段	含义
code	SQLSTATE 错误码
detail	详细信息
constraint	违反的约束名
table / column	相关表与列
severity	级别

常见错误码：

错误码	含义	处理
23505	唯一约束冲突	转为 409，或用 ON CONFLICT 幂等
23503	外键约束失败	校验关联数据
23502	非空约束失败	参数校验
22P02	类型转换失败（如把 "abc" 传给 uuid）	校验入参类型
40001	串行化失败	重试事务
40P01	死锁	重试，并检查加锁顺序
57014	语句被取消（超时）	优化查询或调整超时
53300	连接数已满	缩小连接池或扩容

function isRetryable(error) {
	return ["40001", "40P01", "53300"].includes(error.code);
}

async function withRetry(task, retries = 3) {
	let lastError;
	for (let attempt = 0; attempt < retries; attempt += 1) {
		try {
			return await task();
		} catch (error) {
			lastError = error;
			if (!isRetryable(error)) throw error;
			await new Promise(resolve => setTimeout(resolve, 50 * 2 ** attempt));
		}
	}
	throw lastError;
}

幂等写入：

INSERT INTO users (email, name) VALUES ($1, $2)
ON CONFLICT (email) DO UPDATE SET name = EXCLUDED.name
RETURNING id;

用 UPSERT 替代"先查再插"，可以避免并发下的唯一冲突。

---

八、连接池与性能

8.1 池大小怎么定

单实例连接数 ≈ CPU 核数 × 2 + 磁盘并发
所有实例连接数之和 < max_connections × 0.7

例子：数据库 max_connections = 100，应用部署 4 个副本，每个副本池上限建议 15 以内。

池过大的问题：数据库为每个连接分配内存与进程/线程，连接数过多反而变慢；池过小则请求排队。

8.2 超时配置

const pool = new Pool({
	connectionString: process.env.DATABASE_URL,
	max: 10,
	connectionTimeoutMillis: 2000, // 等连接的超时
	idleTimeoutMillis: 30000, // 空闲回收
	statement_timeout: 5000, // 单条 SQL 执行超时
	query_timeout: 5000, // 客户端侧超时
	keepAlive: true
});

没有超时的查询会长期占用连接，遇到慢查询时会把整个池拖死。

8.3 观察池状态

console.log({
	total: pool.totalCount, // 已建立的连接数
	idle: pool.idleCount, // 空闲连接数
	waiting: pool.waitingCount // 等待连接的请求数
});

waitingCount 持续大于 0，说明池不够或存在慢查询。

数据库侧：

SELECT count(*) FROM pg_stat_activity WHERE datname = 'appdb';
SELECT pid, state, wait_event, query_start, left(query, 80)
FROM pg_stat_activity WHERE state <> 'idle' ORDER BY query_start;

---

九、流式查询与大结果集

const { pool } = require("./db");

async function streamUsers(onRow) {
	const client = await pool.connect();
	try {
		// 游标：服务端分批返回，避免一次性加载到内存
		const cursor = client.query(
			new (require("pg-cursor"))("SELECT id, name FROM users WHERE status = $1", ["active"], { batchSize: 1000 })
		);

		const { promisify } = require("node:util");
		const read = promisify(cursor.read.bind(cursor));
		let rows = await read(1000);
		while (rows.length > 0) {
			for (const row of rows) onRow(row);
			rows = await read(1000);
		}
	} finally {
		client.release();
	}
}

需要 npm i pg-cursor。没有游标时，大结果集会一次性进内存，几百 MB 的数据足以让进程 OOM。

分页导出也可以用 LIMIT/OFFSET 或 keyset 分页（WHERE id > $1 ORDER BY id LIMIT 1000），实现更简单。

---

十、LISTEN / NOTIFY

适合跨实例的轻量事件通知。

const { Client } = require("pg");

async function listenChanges() {
	const client = new Client({ connectionString: process.env.DATABASE_URL });
	await client.connect(); // LISTEN 必须用长连接 Client，不能用 Pool

	client.on("notification", message => {
		console.log("收到通知", message.channel, message.payload);
	});

	await client.query("LISTEN user_created");
}

async function notify(payload) {
	await pool.query("SELECT pg_notify($1, $2)", ["user_created", JSON.stringify(payload)]);
}

注意：LISTEN 需要专用连接且 Pool 会归还连接导致监听失效；断线重连后要重新 LISTEN；通知不保证送达，不能当消息队列用。

---

十一、优雅关闭与健康检查

process.on("SIGTERM", async () => {
	console.log("正在关闭");
	await pool.end(); // 等待已借出的连接归还后关闭
	process.exit(0);
});

健康检查：

async function checkDatabase() {
	try {
		await pool.query("SELECT 1");
		return true;
	} catch {
		return false;
	}
}

健康检查应设置超时，避免探活请求本身被慢查询拖住。

---

十二、ORM 与 Query Builder 概览

方案	迁移	类型安全	学习成本	适合
pg + 手写 SQL	自己管理	需手写类型	低	简单服务、复杂 SQL
Prisma	prisma migrate	强（生成客户端）	中	新项目、团队协作
TypeORM	migration:generate	中	中	NestJS 生态
Drizzle	drizzle-kit	强	低	喜欢 SQL 风格
Knex	knex migrate	弱	低	动态查询

使用 ORM 的两个注意点：

1. N+1 查询：关系加载默认可能逐条查询，用 include/with/join 显式预加载。
2. 逃生通道：复杂报表直接用原生 SQL（prisma.$queryRaw、repository.query），不要硬套 ORM 语法。

---

十三、测试与本地环境

# 本地起库
docker run -d --name pg -p 5432:5432 \
  -e POSTGRES_USER=app -e POSTGRES_PASSWORD=secret -e POSTGRES_DB=appdb \
  postgres:16-alpine

测试策略：

• 单元测试：mock 数据访问层，不连数据库。
• 集成测试：用真实数据库（容器或内存实例），每个测试用例跑在事务里并回滚，保证隔离。

beforeEach(async () => {
	client = await pool.connect();
	await client.query("BEGIN");
});

afterEach(async () => {
	await client.query("ROLLBACK");
	client.release();
});

---

十四、常见问题

现象	原因	处理
password authentication failed	连接串密码错，或 pg_hba.conf 不允许	核对连接串与认证方式
ECONNREFUSED	端口/主机错，容器内用了 localhost	容器内用服务名
timeout exceeded when trying to connect	池耗尽或数据库慢	查 pool.waitingCount 与 pg_stat_activity
relation "users" does not exist	search_path 或 schema 不对	指定 schema：public.users
invalid input syntax for type uuid	入参不是合法 UUID	用 ParseUUIDPipe 或正则校验
金额出现小数误差	numeric 被转成了 float	保持字符串或用 decimal 库
时间差 8 小时	列用了 timestamp 或时区配置不一致	用 timestamptz，统一时区
进程退出时连接未释放	未调用 pool.end()	监听 SIGTERM 并关闭
事务不生效	用了 pool.query 而不是同一个 client	事务内统一使用 client.query

---

十五、FAQ

Q1：pool.query() 和 client.query() 怎么选？

单条独立查询用 pool.query()（自动获取并归还连接）；事务、LISTEN、COPY 用 client.query() 并手动 release()。

Q2：为什么 rowCount 是 null？

部分 SELECT 或某些版本的驱动不返回 rowCount，用 rows.length 判断。

Q3：bigint / numeric 为什么是字符串？

避免超过 Number.MAX_SAFE_INTEGER 或浮点精度问题。需要数字时显式转换，并确认范围。

Q4：如何防止 SQL 注入？

全部使用参数化（$1），表名/列名白名单校验，禁止字符串拼接用户输入。

Q5：连接池应该开多大？

从 CPU 核数 × 2 + 1 起步，压测后调整；重点是所有实例的总连接数不超过数据库上限的 70%。

Q6：事务里能做 HTTP 请求吗？

不要。外部请求耗时会拉长事务、持有锁，失败还会导致回滚。拆成"先请求、后写库"或补偿流程。

Q7：如何做分页？

浅分页用 LIMIT/OFFSET；深分页用 keyset（WHERE id > lastId ORDER BY id LIMIT n）。

Q8：ON CONFLICT 和先查询再插入哪个好？

并发下 ON CONFLICT 更可靠，一次往返且无竞态。

Q9：如何监控慢查询？

数据库开启 log_min_duration_statement 与 pg_stat_statements；应用侧记录查询耗时并打点。

Q10：ESM 项目怎么用？

import pg from "pg";
const { Pool } = pg; // pg 是 CommonJS 包，默认导入后解构

---

附录：API 速查

const { Pool, Client, types } = require("pg");

// 连接池
const pool = new Pool({ connectionString, max: 10, statement_timeout: 5000 });
pool.totalCount; pool.idleCount; pool.waitingCount;
pool.on("error", handler);
await pool.end();

// 查询
await pool.query("SELECT * FROM t WHERE id = $1", [id]);       // 参数化
await pool.query({ name: "q1", text: "SELECT $1::int", values: [1] }); // 预处理

// 事务
const client = await pool.connect();
try { await client.query("BEGIN"); /* ... */ await client.query("COMMIT"); }
catch (e) { await client.query("ROLLBACK"); throw e; }
finally { client.release(); }

// 结果
result.rows; result.rowCount; result.fields;

// 类型解析
types.setTypeParser(1700, v => Number(v)); // numeric
types.setTypeParser(20, v => Number(v));   // bigint（注意范围）

// 长连接（LISTEN / COPY）
const listener = new Client({ connectionString });
await listener.connect();
await listener.query("LISTEN channel");
listener.on("notification", msg => {});