一、从排队说起
想象一下，你去银行办业务，前面只有一条队伍。这叫队列。程序里的队列，本质上也是只能从一头进、另一头出，先来后到，公平有序。可是现实生活里，如果好几个人同时插队、窗口又多，队伍就乱了。程序里的队列在“多个人同时操作”的时候，也会乱套。这就是我们要聊的并发访问控制，以及数据一致性。
这里“多个人”就是多个线程或者多个进程，“同时操作”就是并发访问。只要大家都想从队列里拿东西、放东西，就可能会打架。打架的结果，轻则拿错东西，重则整个程序崩掉。别觉得这是小事，很多线上事故就是这么来的。
二、并发一多，问题就来了
再比如说，一个队伍里突然有两个窗口同时喊“下一位，请过来”。下一位只有一个，两个窗口却都以为自己叫到的是同一个人。结果下一位左右为难，甚至可能被两个人同时拉住。程序里的队列遇到并发访问，就是这种尴尬局面。
2.1 两个客户同时取号
假设咱们有一个最简单的待办任务队列，里面存了几个任务，多个程序同时来取任务。每个任务只能被一个人拿走，就像只有一个苹果，两个人一起抢，最后必须有一个人拿到，另一个空手。但是问题在于“检查队列是否空”和“取出任务”这两步并不是天生的连着一起的。
你可以想象这样的场景：两个小偷同时去偷一个保险箱。A先打开柜子看了一眼，发现有钱，正准备伸手；B也打开柜子看了一眼，发现还有钱，也伸手。最后两个人可能各自拿走了一半，或者一个觉得钱还在，结果被另一个拿走了。程序里的队列数据就因此不一致了。
2.2 一个简单的示例
我们先用代码演示一下这个“打架”的过程。这里统一用 Node.js 的 JavaScript 来写。为了模拟多个客户端同时处理任务，我们用几个并发请求。
// 技术栈：JavaScript（Node.js）
// 模拟一个任务队列，里面放着5个任务
const taskQueue = [1, 2, 3, 4, 5];
// 模拟从队列中取任务，返回取到的任务，队列空了返回null
function takeTask(taskId) {
// 注意：在真实的异步操作中，这里会有个耗时，
// 比如读取数据库、调用远程服务等。
// 我们用一个setTimeout模拟这个耗时。
return new Promise((resolve) => {
setTimeout(() => {
// 关键动作：先看看队列还有没有任务
if (taskQueue.length > 0) {
// 有任务就把它取出来（shift会从头部删除并返回）
const task = taskQueue.shift();
console.log(`取任务的人${taskId} 拿到了任务 ${task}`);
resolve(task);
} else {
console.log(`取任务的人${taskId} 发现队列空了`);
resolve(null);
}
}, Math.random() * 100); // 随机耗时，模拟并发交错
});
}
// 同时有5个人来取任务
async function join() {
const persons = [];
for (let i = 1; i <= 5; i++) {
persons.push(takeTask(i));
}
await Promise.all(persons);
}
join();
// 运行后你就会发现，有人拿到了同一个任务1，
// 或者有人明明看到队列有东西，最后却取不到。
这段代码的问题，相信你已经看出来了。taskQueue.length > 0 和 taskQueue.shift() 不是一个整体。当 A 去检查时，队列确实有任务1，然后他在等待耗时，B 也去检查，队列还没被 A 拿走，所以 B 也认为有任务1。结果 A 执行 shift 拿走任务1，B 也执行 shift，但这时队列可能已经被拿走了，B 拿到的可能是任务2，或者，如果队列里只有一个任务1，B 的 shift 会拿到 undefined。这就是数据不一致。
三、给队列加一把锁
遇到这种问题，最简单粗暴的办法就是“排队进厕所”。先到的人把门锁上，办完事再开门，后面的人等着。这里的锁，就是互斥锁。
3.1 互斥锁怎么用
锁的意思是：同一时间只允许一个人操作队列。其他人在门口排队，等前面的人用完，才能进来。我们可以在代码里加一个简单的互斥锁。为了演示方便，我们使用一个常见的库 async-mutex。当然，你也可以自己写锁，但那个容易写错，直接用现成的更稳。
// 技术栈：JavaScript（Node.js）
// 需要先安装依赖：npm install async-mutex
const { Mutex } = require('async-mutex');
const mutex = new Mutex();
// 模拟一个任务队列
const taskQueue = [1, 2, 3, 4, 5];
// 带锁的取任务函数
async function takeTask(taskId) {
// 用 runExclusive 把整个操作包起来，相当于进门后锁门
await mutex.runExclusive(async () => {
// 模拟真实业务耗时，比如读数据库、网络请求
await new Promise((r) => setTimeout(r, Math.random() * 100));
// 现在整个操作是独占的，不会有人插进来
if (taskQueue.length > 0) {
const task = taskQueue.shift();
console.log(`取任务的人${taskId} 拿到了任务 ${task}`);
} else {
console.log(`取任务的人${taskId} 发现队列空了`);
}
});
}
// 同时5个人来取任务
async function join() {
const persons = [];
for (let i = 1; i <= 5; i++) {
persons.push(takeTask(i));
}
await Promise.all(persons);
}
join();
// 运行结果：每个人都拿到不重复的任务，不会出现两个人抢一个任务的情况
加了锁以后，检查队列和取任务这两步就合成一个原子操作了。就像把保险箱的门锁上，别人必须等你完事才能打开，自然不会出现里面钱被两个人同时看到的混乱。
但是锁也有缺点，如果太多人排队等锁，性能会变差，而且一旦某个拿锁的线程中途卡死，后面的人就得一直等着。我们需要更轻松一点的办法。
四、原子操作：不需要锁的“专家级”操作
有些操作本身就可以保证“要么不执行，要么执行完”，中间不会被插一脚。这就是原子操作。比如很多编程语言里都有“原子的队列操作”，或者专门为并发设计的原子类。在 JavaScript 里，我们可以借助 Atomics 和 SharedArrayBuffer 实现底层原子操作，但更常见的是用消息队列中间件自带的原子能力。
不过为了不引入别的语言，我们这里仍然用 JavaScript 写一个简单的“先比较后取值”的模拟思路。虽然模拟不能完全代表真原子，但意思是一样：我们利用一个“令牌”来保证只有拿到令牌的人才能干活。
// 技术栈：JavaScript（Node.js）
// 这不是一个真正的无锁原子实现，而是演示“原子替换”思想
// 队列中剩余任务数，我们把它放在一个普通对象里
let remaining = 5;
// 做一个“令牌”，谁拿到这个令牌，谁就执行取任务
let token = 0;
async function takeTask(taskId) {
// 尝试拿一个新的令牌
const myToken = token + 1;
token = myToken;
// 模拟一个耗时的检查操作
await new Promise((r) => setTimeout(r, Math.random() * 100));
// 如果当前令牌还是我的，说明我是最后一个进入的，可以执行
if (token === myToken && remaining > 0) {
remaining--;
console.log(`取任务的人${taskId} 拿到了任务 #${remaining + 1}`);
} else {
console.log(`取任务的人${taskId} 放弃了，因为令牌已失效`);
}
}
// 同时5个人来取
async function join() {
const persons = [];
for (let i = 1; i <= 5; i++) {
persons.push(takeTask(i));
}
await Promise.all(persons);
}
join();
// 这个例子只是展示“用令牌抢占”的思想，不代表生产级别的原子实现
这个例子只是让你感受一下“原子”到底是什么。现实中，如果真想无锁操作，推荐使用已经实现好的原子队列，或者使用下面要说的阻塞队列。
五、阻塞队列：把判空和等待交给专业选手
与其自己用锁来实现，不如直接用一个专门为并发设计的“阻塞队列”。阻塞队列的意思是：如果队列空了，取任务的人会主动等待，直到队列有数据；如果队列满了，放任务的人也会等待，直到有位置。这个动作由队列自己控制，不需要你去锁。
很多语言的标准库里都有阻塞队列，比如 Java 的 LinkedBlockingQueue。但是在我们的 JavaScript 示例中，我们可以利用 async-mutex 加“条件变量”来实现类似的等待。条件变量是锁的升级版。它不光让你排队，还会在你等待的时候主动让出 CPU，等条件满足了再把你叫醒。这样就不会傻傻地占用资源。很多阻塞队列内部其实都是靠锁加条件变量实现的。
不过为了避免代码太长，我们用一个更贴近实际的选择：使用 Redis 的阻塞队列。注意，这里只是库不同，语言依然是 JavaScript。
Redis 提供了一个 BRPOPLPUSH 命令，它可以在列表为空时阻塞，并且可以把从队列取出的值备份到另一个列表，防止丢失。我们来看个简单示例。这里需要用到 ioredis 库。
// 技术栈：JavaScript（Node.js）
// 需要先安装依赖：npm install ioredis
const Redis = require('ioredis');
// 模拟两个消费者
const redisA = new Redis();
const redisB = new Redis();
// 任务列表：tasks
// 备份列表：tasks_backup
// 假设 tasks 里已经有5个任务
const TASK_LIST = 'tasks';
const BACKUP_LIST = 'tasks_backup';
async function consume(consumerName) {
// BRPOPLPUSH 从 TASK_LIST 弹出任务，并且写入到 BACKUP_LIST 里
// 如果列表为空，会阻塞等待，直到有任务出现
const task = await redisA.brpoplpush(TASK_LIST, BACKUP_LIST, 0);
// 此时任务已经被我们“暂时拿走”并备份了
console.log(`${consumerName} 拿到任务：${task}`);
// 模拟处理任务，处理成功后从备份列表删除该任务
await new Promise((r) => setTimeout(r, 100));
// 从备份列表中移除已处理的任务
await redisA.lrem(BACKUP_LIST, 1, task);
console.log(`${consumerName} 完成并删除了任务：${task}`);
}
// 两个消费者同时开始消费
consume('消费者A');
consume('消费者B');
// 注意：这里没有关闭Redis连接，因为要保持进程不退出
// 实际项目中请使用 process.on('SIGINT') 优雅关闭
这个示例展示了实际生产环境里常用的一种方式：用 Redis 的原子阻塞命令来保证任务不会被两个人同时取走，同时还能备份。这样即使消费者在处理任务时宕机了，任务还在备份列表里，可以恢复，不会造成数据丢失。
六、什么时候需要关注队列并发
6.1 电商抢购
双十一抢购时，很多人同时把订单放到队列里。如果不好好控制，一个商品可能被卖掉两次。这时候要确保每个商品的数量扣减是原子的。不然你以为库存还剩一百件，结果大家抢到的订单加起来超过二百件，那就麻烦了。
6.2 日志收集
多个服务往同一个日志队列里写日志，如果并发导致日志行错乱，排查问题就会很痛苦。明明是一条完整的错误日志，结果被别的日志从中间插进来，最后你看到的内容拼接在一起，完全没法理解。
6.3 任务调度
后台系统把任务发给多个工人线程，每个任务必须且只能被执行一次。如果并发没控制好，任务可能被重复执行，比如重复发送短信、重复扣款，这都是灾难。用户收到两条扣款短信，心态直接崩了。
6.4 直播弹幕
直播间里的弹幕也是典型的队列场景。几万人同时发弹幕，如果队列处理不好，弹幕顺序会错乱，有些人会看到重复弹幕，有些人会直接丢消息。虽然弹幕丢了不会造成经济损失，但用户体验会变差。
七、锁和队列方案各自的优缺点
7.1 互斥锁
优点是简单、容易理解，适合小规模场景。缺点也很明显：阻塞会让吞吐量下降，而且可能导致死锁。比如两个线程互相等对方释放锁，就僵住了。你等我，我等你，谁也不让谁，程序就卡在那里了。
7.2 原子操作
优点是效率高，不需要阻塞，适合高性能场景。缺点是实现困难，处理不好容易出错，而且不一定所有操作都能原子化。有些复杂业务逻辑根本不是一两个原子方法能搞定的。
7.3 阻塞队列
优点是把复杂的并发控制封装起来，用起来最省心。很多语言和中间件都提供了现成的实现，你只管往里面丢任务、取任务。缺点是依赖特定库或者中间件，系统复杂度增加。你不仅要会写业务代码，还得会部署和运维这些中间件。
八、实践中要注意的坑
第一，不要对队列的普通方法抱有太高的并发信任。比如 JavaScript 数组的 push 和 shift 本身很快，但如果你把判断和操作分开，在多个异步任务里就很可能出错。别以为“都是单线程，应该没事”，异步交错会打破你的幻觉。
第二，加锁的粒度要合理。如果是整个队列操作都加锁，没问题；但如果只是部分操作加锁，仍然会有漏洞。比如你取了任务，但是没把“取任务”和“更新数据库状态”放在同一个锁里，还是可能重复消费。锁一定要覆盖住完整的关键路径。
第三，要考虑超时和重试。如果消费者拿到任务后崩溃了，任务不能丢。可以像上面的 Redis 示例那样，把任务先放到备份列表里，处理成功再删除，否则要重新投递。没有重试机制，一次处理失败就永远丢失，那问题就大了。
第四，不要以为单线程就没有并发问题。现在的 Node.js 虽然是单线程，但当你同时发起多个异步任务时，它们会在事件循环里交错执行，如果遇到 await，仍然会出现阻塞检查点。所以，刚才的示例都是异步并发，不是多线程，但问题一样存在。
第五，消费者要保证幂等。就算你做了所有控制，网络抖动或者中间件重试，也可能导致同一个任务被投递两次。处理任务时带上一个唯一 ID，处理完成记录下来，下次遇到重复 ID 直接跳过。这也是数据一致性保障里非常关键的一环。
九、总结
队列是一个好东西，但并发访问是个大坑。要保证数据一致性，光靠直觉是不够的，必须考虑原子性，用好锁、原子操作或者阻塞队列。在实际项目中，优先选择成熟的队列中间件，比如 Redis、RabbitMQ、Kafka 等，它们已经替我们做好了并发控制。如果你只需要在单个进程里用队列，也要记得用锁或者专业并发容器。
说到底，队列的并发访问控制不是某个高端框架特有的功能，而是一种编程思维。你要时刻问自己：我在判断一个条件和使用这个条件之间，会不会被其他人打断？如果会，就得想办法把这两个动作绑定在一起。锁也好，原子操作也好，阻塞队列也好，都是绑定动作的工具。选择哪个工具，取决于你的场景是追求简单、追求性能，还是追求可靠。希望这篇文章能让你在写队列代码的时候多留个心眼，少踩几个并发的大坑。
Comments