共享状态、锁与消息队列

共享状态、锁与消息队列

JS/TS 开发者通常把状态放在对象或闭包里，event loop 让许多代码暂时不必面对同一时刻的线程写入。进入 Python 线程、C++ 多线程或机器人回调后，多个执行单元同时读写同一对象会产生竞态；队列和不可变消息往往比到处加锁更容易验证。即使 Python 有 GIL，也不能把“读—改—写”业务不变量当成自动原子操作。

先明确共享状态的不变量，例如 latest 必须是完整的 Reading，不能让消费者读到一半更新的数据。再决定由锁保护、由单一拥有者串行处理，还是通过消息复制。锁减少复制但增加等待和死锁风险；消息隔离更清晰，却需要处理队列容量、延迟和丢弃策略。

学习目标

• 能区分共享状态、消息所有权和临界区，并识别数据竞争与死锁风险。
• 能选择队列满时阻塞、丢弃或覆盖的策略，并说明它是否符合业务需求。
• 能验证队列容量、关闭路径和背压指标，而不是只测试正常吞吐。

共享状态越少，系统越容易推理

下面这段代码只保留同一个意图，重点观察输入边界、数据流和失败语义，而不是逐字符翻译。

let latest: Reading | undefined;
function onReading(reading: Reading) {
latest = reading;
}

# Python
latest = None
lock = threading.Lock()
with lock:
  latest = reading

// C++
std::mutex mutex;
std::lock_guard guard(mutex);
latest = reading;

锁保护的是不变量

锁的范围应覆盖完整的读—校验—写操作，而不是只包住赋值的一行。Python 使用 threading.Lock，C++ 使用 std::mutex 和 RAII guard；两者都要避免在持锁期间调用未知的网络或用户回调。锁顺序必须固定，超时和关闭路径要释放锁。对只读快照，可以在锁内复制一份后立即解锁，让耗时推理在锁外执行。

std::optional<Reading> Latest::get() const {
  std::lock_guard<std::mutex> guard(mutex_);
  return latest_; // 返回值副本，调用方不持有内部锁或引用
}

void Latest::set(Reading reading) {
  std::lock_guard<std::mutex> guard(mutex_);
  latest_ = std::move(reading);
}

队列、背压与消息所有权

有界队列把生产速度和消费速度的差异显式化：满时可以阻塞、丢弃最旧帧、丢弃新帧或返回错误。实时传感器常保留最新值，审计事件则通常不能丢。消息最好是不可变快照，消费者拥有出队后的对象；不要把一个可变缓冲区同时交给生产者和消费者。

常见错误与排错思路

常见错误是只给写操作加锁，读取多字段时仍看到不一致状态；另一个是持锁等待队列或网络导致整体卡住。排错时用线程 sanitizer、记录锁等待时长和队列深度，给每条消息加 sequence；若结果偶发错乱，先把共享状态改为单线程队列验证逻辑，再恢复并发。死锁则检查锁顺序和异常/取消路径是否释放资源。

先明确有界队列的满载策略

实时传感器常更关心最新值，容量满时可丢弃最旧帧；账务或必须完整处理的任务则不能静默丢数据，通常应背压生产者并设置等待上限。下面用同步数组只演示“保留最新两条”的策略；多线程实现必须用锁、线程安全队列或单一消费者事件循环保证检查与更新是原子操作。

const capacity = 2;
const queue = [];
let dropped = 0;
function pushLatest(item) {
  if (queue.length === capacity) { queue.shift(); dropped += 1; }
  queue.push(item);
}
pushLatest("frame-1"); pushLatest("frame-2"); pushLatest("frame-3");
console.log(queue, dropped); // ["frame-2", "frame-3"] 1

这个示例的删除策略适合可丢弃的最新状态，不适合事件日志、交易或训练样本。队列容量应根据消费延迟、单项大小和可用内存估算，并以高水位、丢弃计数和等待时长验证是否需要调整。

关闭、取消与并发验证

测试空队列读取、满队列写入、生产者退出、消费者异常和系统关闭。关闭信号要让等待中的一侧及时醒来；持锁时不要等待网络或无界队列，否则其他线程无法推进。记录 sequence 与时间戳可以发现乱序、重复和丢失，但要避免把共享计数器本身变成新的竞态。

把每条测试的预期状态写清：例如容量满且策略为“覆盖最旧”时，队列长度保持不变、丢弃计数加一、保留的 sequence 连续递增。这样比只检查“线程最后退出了”更能证明同步协议符合实时数据需求。

若 C++ 出现偶发损坏，使用 ThreadSanitizer 并缩小共享可变状态；Python 有 GIL 也不能保证跨多个步骤的业务不变量原子。若 JS 单线程事件循环中出现交错，检查 await 前后共享数据是否已被其他回调修改。

迁移练习

请完成：设计一个传感器生产者和推理消费者之间的有界消息队列，选择容量、满队列策略、消息所有权和关闭信号，并写出如何观察背压。

实时传感器可用固定容量队列并保留最新帧，记录 dropped_total 和 queue_depth；每条消息是不可变快照。消费者停止时关闭队列，让生产者收到 closed 而退出；若消息不可丢，则阻塞生产者并设置最大等待时间。

小结

同步首先是所有权和不变量问题，再是锁 API 选择。明确消息能否丢、队列多大、谁负责关闭，并用背压指标验证；下一步任务调度会在此基础上选择执行单元和取消语义。