Node.js 操作 Redis

2026/11/13 · by ZHANG XING YUN · 阅读 1 次 · 3798 字 · 预计阅读 10 分钟

本文讲在 Node.js 里如何使用 Redis：客户端选型、连接与重连、常用数据结构操作、缓存与序列化、分布式锁、限流、管道与事务、Lua 脚本、发布订阅与 Stream，以及生产环境的连接管理与监控。

环境：Node.js 20+、Redis 7、ioredis 5.x、redis（node-redis）4.x/5.x
示例以 ioredis 为主，同时给出 node-redis 的对应写法

一、客户端选型

方案	特点	适用
`ioredis`	API 完整、支持 Sentinel/Cluster、Lua 定义、管道与事务好用	大多数服务端项目
`redis`（node-redis）	官方维护、v4 起全面 Promise 化、类型定义好	偏好官方客户端
其他封装（cache-manager 等）	提供统一缓存接口	需要切换缓存后端

选择建议：

需要 Cluster / Sentinel、复杂 Lua、scanStream：ioredis。
只用基础命令、希望与官方文档一致：node-redis。

两者不要在同一进程中混用同一连接配置，重连与命令队列策略不同，排查困难。

二、安装与连接

npm i ioredis

2.1 基础连接

// redis.js
const Redis = require("ioredis");

const redis = new Redis(process.env.REDIS_URL, {
	maxRetriesPerRequest: 3, // 单条命令最多重试次数，避免请求无限挂起
	enableReadyCheck: true, // 连接后确认服务端就绪
	connectTimeout: 5000,
	commandTimeout: 2000, // 单条命令超时（毫秒）
	retryStrategy(times) {
		// 重连间隔：最多 2 秒
		return Math.min(times * 100, 2000);
	}
});

redis.on("connect", () => console.log("Redis 已连接"));
redis.on("ready", () => console.log("Redis 可用"));
redis.on("error", error => console.error("Redis 错误", error.message));

module.exports = { redis };

关键配置：

配置	说明	建议
`maxRetriesPerRequest`	单命令重试次数	3；缓存场景可更小，避免拖慢接口
`enableReadyCheck`	连接后检查 `INFO`	保持 `true`
`connectTimeout`	建立连接超时	5000ms
`commandTimeout`	单命令超时	1000~3000ms
`retryStrategy`	重连退避	指数退避，上限 2s
`lazyConnect`	延迟连接	需要先注册监听时用
`keyPrefix`	自动给 key 加前缀	多业务共用实例时用

2.2 node-redis 写法

const { createClient } = require("redis");

const client = createClient({ url: process.env.REDIS_URL });
client.on("error", error => console.error("Redis 错误", error.message));

await client.connect(); // v4 起必须显式连接

三、基本操作与两种客户端对照

操作	ioredis	node-redis
设置	`await redis.set("k", "v")`	`await client.set("k", "v")`
带过期	`await redis.set("k", "v", "EX", 60)`	`await client.set("k", "v", { EX: 60 })`
不存在才设	`await redis.set("k", "v", "NX", "EX", 10)`	`await client.set("k", "v", { NX: true, EX: 10 })`
读取	`await redis.get("k")`	`await client.get("k")`
自增	`await redis.incr("n")`	`await client.incr("n")`
哈希	`await redis.hset("h", { a: 1 })`	`await client.hSet("h", { a: 1 })`
列表	`await redis.lpush("l", "a")`	`await client.lPush("l", "a")`
集合	`await redis.sadd("s", "a")`	`await client.sAdd("s", "a")`
有序集合	`await redis.zadd("z", 1, "a")`	`await client.zAdd("z", { score: 1, value: "a" })`
删除	`await redis.del("k")`	`await client.del("k")`
批量删除	`await redis.del("a", "b")`	`await client.del(["a", "b"])`
扫描	`redis.scanStream()`	`client.scanIterator()`

常用结构示例：

// String：计数器
const count = await redis.incr("visits:2026-01-13");

// Hash：对象字段
await redis.hset("user:1", { name: "张三", age: 18 });
const user = await redis.hgetall("user:1");

// List：队列
await redis.lpush("queue:mail", JSON.stringify({ to: "a@example.com" }));
const task = await redis.rpop("queue:mail");

// Set：去重
await redis.sadd("online:users", "1001");
await redis.sismember("online:users", "1001");

// ZSet：排行榜
await redis.zadd("rank", 100, "user:1");
await redis.zincrby("rank", 50, "user:1");
const top10 = await redis.zrevrange("rank", 0, 9, "WITHSCORES");

返回值约定（与 redis-cli 一致）：

命令	未命中时返回
`GET`	`null`
`HGETALL`	`{}`
`ZSCORE`	`null`
`EXISTS`	`0`
`TTL`	`-2`（不存在）/ `-1`（无过期）
`SET ... NX` 失败	`null`

四、序列化约定

Redis 只存字节，对象必须序列化。

// 统一封装，避免各处写法不一致
async function setJson(key, value, ttlSeconds) {
	const payload = JSON.stringify(value);
	if (ttlSeconds) {
		await redis.set(key, payload, "EX", ttlSeconds);
	} else {
		await redis.set(key, payload);
	}
}

async function getJson(key) {
	const raw = await redis.get(key);
	if (raw === null) return null;
	try {
		return JSON.parse(raw);
	} catch {
		await redis.del(key); // 脏数据直接删除，避免反复报错
		return null;
	}
}

约定建议：

缓存对象一律 JSON，字段变更时升级 key 版本（user:v2:1001），避免新旧结构混用。
不要把 undefined 放进对象（JSON.stringify 会丢掉字段）。
二进制数据（图片、protobuf）用 Buffer，不要转 JSON。
金额不要存浮点数，存字符串或最小货币单位整数。

五、缓存模式（Cache-Aside）完整实现

// cache.js
const crypto = require("node:crypto");
const { redis } = require("./redis");

const DEFAULT_TTL = 300;

async function getOrLoad(key, loader, options = {}) {
	const ttl = options.ttl ?? DEFAULT_TTL;

	const cached = await redis.get(key);
	if (cached !== null) {
		return { value: JSON.parse(cached), source: "cache" };
	}

	// 互斥锁：热点 key 失效时只放一个请求回源数据库
	const lockKey = `lock:${key}`;
	const token = crypto.randomUUID();
	const acquired = await redis.set(lockKey, token, "PX", 3000, "NX");

	if (!acquired) {
		// 没抢到锁：短暂等待后重试读缓存，避免所有请求同时打数据库
		await new Promise(resolve => setTimeout(resolve, 50));
		const retry = await redis.get(key);
		if (retry !== null) {
			return { value: JSON.parse(retry), source: "cache-retry" };
		}
		return { value: await loader(), source: "db-fallback" };
	}

	try {
		const value = await loader();

		// TTL 加随机抖动，避免大量 key 同时过期（缓存雪崩）
		const jitter = Math.floor(ttl * 0.1 * Math.random());
		await redis.set(key, JSON.stringify(value), "EX", ttl + jitter);

		return { value, source: "db" };
	} finally {
		// 只释放自己持有的锁
		await redis.eval(
			"if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end",
			1,
			lockKey,
			token
		);
	}
}

async function invalidate(key) {
	await redis.del(key);
}

module.exports = { getOrLoad, invalidate };

使用：

const { getOrLoad, invalidate } = require("./cache");

const { value: user, source } = await getOrLoad(
	`user:v1:${id}`,
	() => usersService.findById(id),
	{ ttl: 600 }
);

// 更新数据库后删缓存，而不是改缓存
await usersService.update(id, patch);
await invalidate(`user:v1:${id}`);

三个要点：

先更新数据库，再删除缓存；删除失败要有重试或异步补偿。
缓存空值防穿透：查不到时存 "null" 并给短 TTL（如 30 秒）。
TTL 加抖动，避免同一时刻批量失效。

六、分布式锁

// lock.js
const crypto = require("node:crypto");
const { redis } = require("./redis");

const RELEASE_LUA = `
if redis.call("get", KEYS[1]) == ARGV[1] then
  return redis.call("del", KEYS[1])
else
  return 0
end`;

async function acquireLock(key, ttlMs = 5000) {
	const token = crypto.randomUUID();
	const ok = await redis.set(key, token, "PX", ttlMs, "NX");
	return ok === "OK" ? { key, token } : null;
}

async function releaseLock(lock) {
	await redis.eval(RELEASE_LUA, 1, lock.key, lock.token);
}

// 需要更长执行时间时续期（watchdog）
async function extendLock(lock, ttlMs = 5000) {
	const result = await redis.eval(
		`if redis.call("get", KEYS[1]) == ARGV[1] then
		   return redis.call("pexpire", KEYS[1], ARGV[2])
		 else return 0 end`,
		1,
		lock.key,
		lock.token,
		ttlMs
	);
	return result === 1;
}

module.exports = { acquireLock, releaseLock, extendLock };

使用：

const { acquireLock, releaseLock } = require("./lock");

const lock = await acquireLock(`lock:order:${orderId}`, 10000);
if (!lock) {
	throw new Error("操作进行中，请稍后重试");
}
try {
	await doSomething();
} finally {
	await releaseLock(lock);
}

必须满足：

加锁与设置过期是同一条命令（SET ... NX PX），不能分两步。
释放时校验 token，否则可能删掉别人的锁。
过期时间大于业务最长耗时，否则锁提前失效；不确定时用续期。

局限：Redis 主从是异步复制，主节点宕机可能丢锁。对账务等强一致场景，应使用数据库行锁或分布式协调服务，而不是只依赖 Redis 锁。

七、限流

7.1 固定窗口

async function fixedWindowLimit(ip, limit = 60, windowSeconds = 60) {
	const window = Math.floor(Date.now() / (windowSeconds * 1000));
	const key = `rate:${ip}:${window}`;

	const count = await redis.incr(key);
	if (count === 1) {
		await redis.expire(key, windowSeconds);
	}
	return { allowed: count <= limit, count };
}

实现简单，但窗口边界会有双倍突发。

7.2 滑动窗口（ZSet）

async function slidingWindowLimit(key, limit = 60, windowMs = 60000) {
	const now = Date.now();
	const member = `${now}-${Math.random()}`;

	const pipeline = redis.pipeline();
	pipeline.zadd(key, now, member);
	pipeline.zremrangebyscore(key, 0, now - windowMs);
	pipeline.zcard(key);
	pipeline.pexpire(key, windowMs);

	const results = await pipeline.exec();
	const count = results[2][1];
	return { allowed: count <= limit, count };
}

7.3 令牌桶（Lua 实现）

redis.defineCommand("tokenBucket", {
	numberOfKeys: 1,
	lua: `
	local key = KEYS[1]
	local capacity = tonumber(ARGV[1])
	local rate = tonumber(ARGV[2])
	local now = tonumber(ARGV[3])
	local state = redis.call("hmget", key, "tokens", "ts")
	local tokens = tonumber(state[1]) or capacity
	local ts = tonumber(state[2]) or now
	tokens = math.min(capacity, tokens + (now - ts) / 1000 * rate)
	local allowed = 0
	if tokens >= 1 then
	  tokens = tokens - 1
	  allowed = 1
	end
	redis.call("hmset", key, "tokens", tokens, "ts", now)
	redis.call("pexpire", key, 3600000)
	return allowed
	`
});

const allowed = await redis.tokenBucket(`bucket:${ip}`, 10, 5, Date.now());

八、管道与事务

8.1 管道（pipeline）

const pipeline = redis.pipeline();
pipeline.set("a", 1);
pipeline.incr("a");
pipeline.expire("a", 60);

const results = await pipeline.exec();
// [[null, "OK"], [null, 2], [null, true]]

exec() 返回 [error, result] 数组，必须逐项检查 error。

两种客户端的返回值结构不同：ioredis 的 exec() 返回 [[err, res], ...]，node-redis 的 multi().exec() 直接返回结果值数组（如 ["OK", 2]），错误以异常形式抛出。

8.2 事务（multi）

const results = await redis
	.multi()
	.set("balance", 100)
	.decrby("balance", 10)
	.exec();

multi 保证命令连续执行不被插入，但不支持回滚：某条命令运行时报错，其它命令仍生效。

8.3 带条件的原子操作

// 仅当值等于期望时才更新（乐观锁）
const updated = await redis.eval(
	`if redis.call("get", KEYS[1]) == ARGV[1] then
	   return redis.call("set", KEYS[1], ARGV[2])
	 else return 0 end`,
	1,
	"config:mode",
	"v1",
	"v2"
);

九、Lua 脚本

redis.defineCommand("releaseLock", {
	numberOfKeys: 1,
	lua: `
	if redis.call("get", KEYS[1]) == ARGV[1] then
	  return redis.call("del", KEYS[1])
	else
	  return 0
	end`
});

// 调用：参数顺序为 key、然后普通参数
await redis.releaseLock("lock:1", token);

要点：

脚本执行是原子的，适合"读-判断-写"。
脚本内不要写死 key（集群模式下所有 key 需在同一槽）。
脚本要短，单次执行控制在毫秒级，否则阻塞其它命令。
用 defineCommand 而非每次 eval 字符串，便于复用与测试。

十、发布订阅

const { redis } = require("./redis");

// 订阅需要独立连接（订阅模式下不能执行普通命令）
const subscriber = redis.duplicate();
await subscriber.subscribe("order.created");
subscriber.on("message", (channel, message) => {
	console.log(channel, JSON.parse(message));
});

// 发布
await redis.publish("order.created", JSON.stringify({ orderId: 1001 }));

特点：fire-and-forget，订阅者离线期间的消息不会补发；需要可靠投递用 Stream。

十一、Stream 消息队列

// 生产
await redis.xadd("events", "*", "type", "order.created", "orderId", "1001");

// 创建消费组（不存在时）
try {
	// 最后一个参数是起始位置：$ 表示只消费创建之后的新消息，0 表示从最早的消息开始
	await redis.xgroup("CREATE", "events", "orders", "$", "MKSTREAM");
} catch (error) {
	if (!String(error.message).includes("BUSYGROUP")) throw error;
}

// 消费
async function consume() {
	const results = await redis.xreadgroup(
		"GROUP", "orders", "consumer-1",
		"COUNT", 10,
		"BLOCK", 5000,
		"STREAMS", "events", ">"
	);
	// 没有新消息时返回 null，属于正常情况
	if (!results) return;

	for (const [, messages] of results) {
		for (const [id, fields] of messages) {
			try {
				console.log(id, fields);
				await redis.xack("events", "orders", id);
			} catch (error) {
				// 未确认的消息会进入 pending，可后续用 XPENDING / XCLAIM 处理
				console.error("处理失败", id, error);
			}
		}
	}
}

消费组提供确认与重投，适合轻量事件流；需要复杂路由、持久化与高吞吐时使用 Kafka/RabbitMQ。

十二、连接管理：重连、集群、哨兵

12.1 重连与降级

const redis = new Redis(url, {
	maxRetriesPerRequest: 1,
	retryStrategy: times => Math.min(times * 200, 3000),
	reconnectOnError: error => error.message.includes("READONLY")
});

缓存不可用时应该降级而不是让接口直接失败：

async function safeGet(key) {
	try {
		return await redis.get(key);
	} catch (error) {
		console.warn("Redis 读取失败，降级到数据库", error.message);
		return null;
	}
}

12.2 哨兵

const redis = new Redis({
	sentinels: [
		{ host: "10.0.0.1", port: 26379 },
		{ host: "10.0.0.2", port: 26379 }
	],
	name: "mymaster",
	sentinelRetryStrategy: times => Math.min(times * 200, 3000)
});

12.3 集群

const Redis = require("ioredis");

const cluster = new Redis.Cluster(
	[
		{ host: "10.0.0.1", port: 6379 },
		{ host: "10.0.0.2", port: 6379 }
	],
	{
		redisOptions: { password: process.env.REDIS_PASSWORD },
		scaleReads: "slave" // 读请求分配到从节点
	}
);

集群限制：多 key 操作要求 key 在同一槽，用 hash tag 归并：

// {user:1} 保证这两个 key 落在同一槽
await cluster.mget("{user:1}:name", "{user:1}:age");

十三、错误处理与监控

redis.on("error", error => {
	// 不要在这里抛错，连接错误会持续触发
	console.error("Redis 连接错误", error.message);
});

redis.on("reconnecting", delay => {
	console.warn(`Redis 重连中，${delay}ms 后重试`);
});

关键指标：

const info = await redis.info("stats");
// keyspace_hits / keyspace_misses 计算命中率
const memory = await redis.info("memory");
// used_memory、mem_fragmentation_ratio

应用侧建议记录：

命令耗时（慢查询）。
错误率与降级次数。
连接状态变化次数（频繁重连说明网络或服务端不稳定）。

排查大 key：

const stream = redis.scanStream({ match: "user:*", count: 200 });
stream.on("data", async keys => {
	for (const key of keys) {
		const type = await redis.type(key);
		const size = await redis.memory("USAGE", key);
		if (size > 1024 * 1024) console.warn("大 key", key, type, size);
	}
});

十四、优雅关闭

process.on("SIGTERM", async () => {
	try {
		await redis.quit(); // 发送 QUIT，等待已发出的命令完成
		process.exit(0);
	} catch (error) {
		redis.disconnect(); // 强制断开
		process.exit(1);
	}
});

区别：

quit()：发送 QUIT，正常关闭，建议使用。
disconnect()：立即断开，未完成的命令会失败。

十五、常见问题

现象	原因	处理
请求偶发变慢	`maxRetriesPerRequest` 过大，命令一直重试	调小并设置 `commandTimeout`
启动后进程不退出	连接未关闭	`SIGTERM` 中调用 `quit()`
`WRONGTYPE`	同一 key 用了不同结构	统一 key 规范，避免前缀冲突
`OOM command not allowed`	达到 `maxmemory` 且策略为 noeviction	调整淘汰策略或扩容
内存不下降	过期 key 未及时回收、碎片	检查 `MEMORY DOCTOR`、`activedefrag`
订阅收不到消息	复用了命令连接	用 `duplicate()` 独立连接
集群多 key 报 `CROSSSLOT`	key 不在同一槽	使用 hash tag
锁提前失效	TTL 小于业务耗时	增大 TTL 或加续期
缓存与数据库不一致	先删缓存后更新库，或删除失败	先更新库再删缓存，失败重试

十六、FAQ

Q1：ioredis 和 node-redis 能混用吗？

不建议。重连策略、命令队列、错误处理都不同，混用会让问题难以定位。

Q2：为什么 set 带 NX 返回 null？

表示 key 已存在，设置未执行。这是分布式锁与幂等控制的正常返回值，不是错误。

Q3：管道和事务的区别？

管道只是批量发送、减少往返，不保证原子；multi 保证命令连续执行，但不支持回滚。

Q4：缓存一定要设 TTL 吗？

要。TTL 是兜底：写缓存逻辑出错、数据源变更时，靠过期让数据最终恢复一致。

Q5：如何避免缓存击穿？

热点 key 用互斥锁重建（见第五节），或逻辑过期 + 异步刷新。

Q6：分布式锁能保证强一致吗？

不能。Redis 主从异步复制，主节点故障可能丢锁。强一致场景用数据库行锁或协调服务。

Q7：scanStream 会阻塞吗？

不会，SCAN 是增量遍历，但可能返回重复 key，需要幂等处理；遍历期间新增的 key 不保证被扫描到。

Q8：Redis 挂了接口怎么办？

缓存层降级：捕获异常后回源数据库，同时加限流保护数据库，避免雪崩。

Q9：连接数开多少？

每个进程一个连接即可（ioredis 内部多路复用）；高并发写场景可开多个连接池，但要注意 Redis 的 maxclients。

Q10：Cluster 下 Lua 脚本有什么限制？

脚本里的所有 key 必须位于同一槽，否则报 CROSSSLOT；用 hash tag 保证。

附录：API 速查

// ioredis
const Redis = require("ioredis");
const redis = new Redis(url, { maxRetriesPerRequest: 3, commandTimeout: 2000 });

await redis.set("k", "v");
await redis.set("k", "v", "EX", 60);
await redis.set("k", "v", "PX", 3000, "NX");
await redis.get("k");
await redis.del("k1", "k2");
await redis.exists("k");
await redis.expire("k", 60);
await redis.ttl("k");
await redis.incr("n");
await redis.hset("h", { a: 1 });
await redis.hgetall("h");
await redis.lpush("l", "a");
await redis.lrange("l", 0, -1);
await redis.sadd("s", "a");
await redis.zadd("z", 1, "a");
await redis.zrevrange("z", 0, 9, "WITHSCORES");
await redis.publish("ch", "msg");

const sub = redis.duplicate();
await sub.subscribe("ch");
sub.on("message", (channel, message) => {});

const pipe = redis.pipeline();
pipe.set("a", 1).incr("a");
await pipe.exec();

await redis.multi().set("a", 1).incr("a").exec();

await redis.eval("return redis.call('get', KEYS[1])", 1, "k");
redis.defineCommand("myCmd", { numberOfKeys: 1, lua: "return 1" });
await redis.myCmd("k", "arg");

await redis.xadd("stream", "*", "f", "v");
await redis.xreadgroup("GROUP", "g", "c", "COUNT", 10, "STREAMS", "stream", ">");
await redis.xack("stream", "g", id);

await redis.quit();
redis.disconnect();

// node-redis
const { createClient } = require("redis");
const client = createClient({ url });
await client.connect();
await client.set("k", "v", { EX: 60, NX: true });
await client.get("k");
await client.multi().set("a", 1).incr("a").exec();
await client.quit();