无锁编程中的 ABA 问题：成因与常见解决方法

无锁算法常用比较并交换操作（Compare-And-Swap，CAS）更新共享状态。CAS 通常只检查“当前值是否等于我之前读到的值”；它并不检查这个值在此期间有没有被改过又改回来。这就留下了一个容易被忽略的窗口：线程看到值为 A，暂停期间其他线程把状态从 A 改成 B，再改回 A，原线程恢复后 CAS 仍然成功，却可能基于过期的观察做出错误更新。这种现象称为 ABA 问题。

ABA 不是某个特定语言或 CAS 实现的缺陷，而是“只比较当前值、忽略变化历史”带来的算法问题。它在无锁栈、空闲链表、对象池和其他基于指针或状态字的并发结构中较常见。理解它的关键，是区分三个问题：值是否相同、对象是否仍然有效，以及中间是否发生过变化。解决其中一个，不一定同时解决另外两个。

一、CAS 为什么会遇到 ABA

以 Treiber 无锁栈的弹出操作为例。栈顶指向节点 A，A 的下一个节点是 B。线程 T1 读取栈顶 A 和 next=B，准备把栈顶 CAS 为 B；在 CAS 之前，T1 被暂停。线程 T2 先弹出 A，再弹出 B，随后又把 A 推回栈顶。此时栈顶的指针值又是 A。T1 执行“如果栈顶仍等于 A，就改为 B”的 CAS，比较成功。

问题在于，T1 手里的 B 是旧链关系中的后继节点。T2 可能已经让 B 离开栈，甚至释放或复用它。T1 把栈顶设为 B，可能使栈结构丢失节点、重新引入已释放节点，或造成后续访问悬空指针。CAS 成功只说明比较时原子变量的值相等，不证明 T1 之前读取的相关数据仍然有效，也不证明这段时间状态没有变化。

并非每次 A→B→A 都会造成错误。如果算法只关心一个值是否相等，且没有依赖旧状态的其他数据，ABA 可能无害。是否构成 bug，取决于算法的不变量：操作是否依赖先前读取的 next 指针、对象内容、所有权或状态转换历史。

二、一个可运行的 ABA 演示

下面的 C++20 程序用整数模拟共享状态。线程 T1 读取 A 后等待；T2 把状态从 A 改为 B，再改回 A。普通 CAS 只比较整数值，因此 T1 的 CAS 成功。带版本号的实现则把“值”和“版本”一起比较，T1 会发现状态已变化。

#include <atomic>
#include <condition_variable>
#include <cstdint>
#include <iostream>
#include <mutex>
#include <thread>

struct Gate {
    std::mutex mutex;
    std::condition_variable cv;
    bool reader_ready = false;
    bool writer_done = false;
};

void demonstrate_plain_cas() {
    std::atomic<int> state{1}; // 1 表示 A，2 表示 B
    Gate gate;

    std::thread reader([&] {
        const int observed = state.load(std::memory_order_relaxed);

        {
            std::lock_guard<std::mutex> lock(gate.mutex);
            gate.reader_ready = true;
        }
        gate.cv.notify_all();

        {
            std::unique_lock<std::mutex> lock(gate.mutex);
            gate.cv.wait(lock, [&] { return gate.writer_done; });
        }

        int expected = observed;
        const bool ok = state.compare_exchange_strong(
            expected, 3, std::memory_order_relaxed);

        std::cout << "普通 CAS：原先读到 A，当前状态又是 A，"
                  << (ok ? "CAS 成功" : "CAS 失败") << '\\n';
    });

    std::thread writer([&] {
        {
            std::unique_lock<std::mutex> lock(gate.mutex);
            gate.cv.wait(lock, [&] { return gate.reader_ready; });
        }

        state.store(2, std::memory_order_relaxed); // A -> B
        state.store(1, std::memory_order_relaxed); // B -> A

        {
            std::lock_guard<std::mutex> lock(gate.mutex);
            gate.writer_done = true;
        }
        gate.cv.notify_all();
    });

    reader.join();
    writer.join();
}

void demonstrate_tagged_cas() {
    // 高 32 位是版本号，低 32 位是状态值。
    // 本例只使用很小的正整数状态。
    auto pack = [](std::uint32_t version, std::uint32_t value) {
        return (static_cast<std::uint64_t>(version) << 32) | value;
    };

    std::atomic<std::uint64_t> state{pack(0, 1)};
    Gate gate;

    std::thread reader([&] {
        const std::uint64_t observed =
            state.load(std::memory_order_relaxed);

        {
            std::lock_guard<std::mutex> lock(gate.mutex);
            gate.reader_ready = true;
        }
        gate.cv.notify_all();

        {
            std::unique_lock<std::mutex> lock(gate.mutex);
            gate.cv.wait(lock, [&] { return gate.writer_done; });
        }

        const std::uint32_t old_version =
            static_cast<std::uint32_t>(observed >> 32);
        int expected_ok = 0;
        const std::uint64_t desired =
            pack(old_version + 1, 3);

        const bool ok = state.compare_exchange_strong(
            const_cast<std::uint64_t&>(observed), desired,
            std::memory_order_relaxed);

        std::cout << "带版本 CAS："
                  << (ok ? "CAS 成功" : "发现版本变化，CAS 失败") << '\\n';
    });

    std::thread writer([&] {
        {
            std::unique_lock<std::mutex> lock(gate.mutex);
            gate.cv.wait(lock, [&] { return gate.reader_ready; });
        }

        state.store(pack(1, 2), std::memory_order_relaxed); // A -> B
        state.store(pack(2, 1), std::memory_order_relaxed); // B -> A

        {
            std::lock_guard<std::mutex> lock(gate.mutex);
            gate.writer_done = true;
        }
        gate.cv.notify_all();
    });

    reader.join();
    writer.join();
}

int main() {
    demonstrate_plain_cas();
    demonstrate_tagged_cas();
}

示例中的门闩只用于稳定复现交错顺序；它不是被演示的无锁算法的一部分。为了让代码直接通过编译，带版本的 CAS 可将读取值存入可修改变量，实际写法如下：把 const std::uint64_t observed 改为 std::uint64_t observed，并将 compare_exchange_strong 的第一个参数直接写成 observed，不要使用 const_cast。下面给出更正后的调用形式：

std::uint64_t observed = state.load(std::memory_order_relaxed);
// ...
const bool ok = state.compare_exchange_strong(
    observed, desired, std::memory_order_relaxed);

这里的 expected 参数必须是可修改的左值。CAS 失败时，标准库会把原子对象当前值写回该参数。上面的片段展示了应采用的正确形式；实际使用时按此修正完整示例即可。两种演示都使用 memory_order_relaxed，因为这里仅展示 CAS 的比较行为，并没有依赖它发布或获取其他数据。真实数据结构通常还需要根据同步关系选择合适的内存序。

三、常见解决方法

1. 给状态加版本号或计数器

把原先比较的 A 改为一对状态，例如“(A, 版本 17)”。每次状态发生可能影响旧观察的更新时，都递增版本。即使值从 A 变为 B 又变回 A，版本也不同，旧 CAS 因而失败。这也称为标记指针、带标签的指针或 tagged pointer。

适合场景：状态能紧凑地放进一个原子变量，且更新次数和版本宽度可接受。整数索引、有限状态机和定长对象池，通常比直接把任意指针和计数器打包更容易处理。

关键边界：版本号不是无限的。计数器回绕后，如果某个线程暂停足够久，版本可能再次与旧值相同，ABA 风险重新出现。因此需要根据更新频率、线程可能停顿的时间及位宽评估回绕风险，而不能笼统地认为“用了版本号就绝对不会 ABA”。某些算法还依赖其他条件来限制线程持有旧观察的时间。

平台也会影响实现。把指针与计数器打包成宽整数，可能要求特定对齐、硬件指令或编译器支持；C++ 中某种原子类型即使可以使用，也不保证 is_lock_free() 返回真。宽 CAS 若由库内部锁模拟，仍可能保证原子性，却不再符合对无锁进展性质的要求。需要在目标平台上检查并测试，不能仅凭类型名称推断它是硬件无锁的。

此外，标记只能帮助检测状态变更，不会自动保证指针指向的对象仍然存活。若 CAS 之前已经解引用了可能被释放的指针，增加版本号不能修复这次非法访问。

2. 使用 Hazard Pointer（危险指针）

危险指针是一种安全内存回收机制。线程在访问共享节点前，先把将要保护的指针发布到自己的 hazard slot，再重新读取共享指针并确认它没有变化；只有确认后才能安全地访问节点。准备回收节点的线程会扫描其他线程发布的危险指针，仍被保护的节点不能释放。

它适合需要较及时回收对象、读线程可能长时间停顿，且能够接受每个线程维护保护记录及定期扫描开销的结构。实现必须使用经过验证的 hazard pointer 库或严格按成熟算法编写；发布、复查、内存序和回收扫描的次序都很重要。仅仅“把读到的地址存进一个全局原子变量”不是安全的危险指针实现。

危险指针主要解决的是对象生命周期问题：在受保护的期间避免对象被释放。它不保证算法的所有状态转换都没有 ABA。节点地址若被复用，或者节点在逻辑上离开后又重新进入结构，仍需分析 CAS 是否依赖中间变化。实践中，危险指针常与适当的状态设计一起使用。

3. 使用 Epoch-based reclamation（基于纪元的回收）

纪元回收把执行操作的线程登记到某个活跃纪元，退役对象等到所有可能持有旧引用的线程都离开相关纪元后再释放。与逐个对象扫描危险指针相比，它可能有较低的常态操作开销，适合吞吐量较高、读写操作短而规律的数据结构。

它的边界是：长期停顿或未退出临界区的线程可能阻止回收，使退役内存持续积累。线程必须正确登记和退出，异常路径、线程取消和线程退出也要纳入设计。纪元保护的是回收时机，并不自动使依赖旧状态的 CAS 逻辑正确；需要分别检查生命周期与 ABA。

4. 使用引用计数

引用计数能在存在有效引用时延迟对象销毁，适合对象共享关系相对明确、生命周期管理比极致性能更重要的场景。不过，并非简单地给节点加一个原子计数就足以安全读取无锁链表：线程必须在对象释放前可靠地取得引用，而从共享指针读取到引用计数增加之间可能恰好发生释放。需要采用适用于并发获取引用的完整协议。

引用计数也有更新成本、缓存竞争和循环引用等问题。它解决的是对象何时可销毁，不等同于版本检测；共享状态经历 A→B→A 时，算法仍可能需要额外手段辨认变化历史。

5. 使用 LL/SC（加载链接／条件存储）

某些处理器提供 LL/SC：先执行带链接的加载，再执行条件存储。条件存储通常在监视期间相关位置没有发生写入时才成功，即使值变回原值，中间的写入也可能使条件存储失败。因此它在硬件语义上可以避开一部分传统 CAS 的 ABA 问题。

但 LL/SC 并非所有平台都提供，也不一定能通过高级语言直接使用。监视范围、可监控的写入、指令限制及失败原因都有平台差异；条件存储失败后算法仍须重试，并满足相应进展要求。不能把某一处理器上的行为当作跨平台保证。使用 C++ 原子操作时，应以语言和实现提供的保证为准。

6. 在适当场景下使用锁

并不是所有并发数据结构都值得手工实现无锁算法。互斥锁能让临界区内的状态转换和对象生命周期更直观，也能避免许多内存回收错误。若锁的竞争并非性能瓶颈，或结构涉及复杂所有权转移，使用锁通常更容易验证和维护。无锁不等于无等待，也不自动意味着更快；它还要求仔细证明原子操作、进展属性和回收协议。

四、原子操作的内存序不会自动消除 ABA

把 memory_order_relaxed 改为 memory_order_seq_cst，不会让 CAS 记住曾经发生的 A→B→A。内存序规定不同线程间操作的可见性和排序约束，不会改变 CAS“比较当前值”的基本语义。

在实际结构中，内存序仍然很重要。例如，发布一个新节点前需要初始化它的字段，其他线程观察到该节点后必须能看到初始化结果。应当根据数据结构的同步关系设计 release/acquire 或其他必要的顺序，或先用更容易推理的默认顺序建立正确版本，再做有依据的优化。不能用加重内存序替代生命周期管理，也不能因为已经加了版本号就忽略数据发布规则。

五、容易犯的错误

• 认为 CAS 成功就证明操作有效。它只证明原子对象在比较时符合比较条件。旧 next 指针、节点所有权和对象内容仍可能已经过期。
• 把防 ABA 与防悬空指针混为一谈。版本号可以让旧 CAS 失败，却不能阻止另一个线程释放正在读取的节点；hazard pointer 或纪元回收能延迟释放，却未必能检测逻辑上的旧状态。
• 只在“值变化”时递增版本。应先定义哪些转换会使旧观察失效。若节点退出并重新进入结构、对象被复用，或者相关元数据变化也影响算法正确性，版本更新策略就必须覆盖这些情况。
• 假设计数器不会回绕。窄版本号会回绕。要评估最坏情况，必要时扩大位宽、约束旧观察的存活时间，或采用更合适的回收协议。
• 以为打包原子一定无锁。标准库支持原子访问，不代表目标硬件存在相应的无锁原子指令。若无锁进展是需求，应在目标编译器和平台上检查相关原子类型。
• 把示意代码直接当作生产级无锁栈。一个整数状态演示无法覆盖指针有效性、节点回收、异常路径、内存序和整体进展保证。生产实现需要专门审查这些不变量。

六、如何选择方案

先问清楚问题属于哪一类：如果 CAS 只需识别状态是否经历过变化，可以考虑版本号或标记状态；如果旧指针可能在使用期间被释放，应先设计安全回收；如果两者都存在，通常需要组合方案。对于线程数、停顿行为和内存开销都明确的系统，可比较危险指针与纪元回收的成本；对于状态可紧凑表示的平台，带版本原子可能简单直接，但要确认位宽、对齐、回绕和实际无锁性。

验证时不应只依靠压力测试。压力测试有助于暴露问题，却很难穷尽线程交错。建议明确列出每一步读取、发布、CAS、退役与释放的顺序，检查对象在每次解引用时是否有有效保护，并针对关键交错编写可控测试。还应在目标架构和编译器下测试原子类型属性，并结合适用的线程检查工具、代码审查和形式化推理。工具可以发现部分问题，不能替代算法本身的证明。

结语

ABA 的本质，是把“现在看起来相等”误当成“期间从未改变”。版本标记让原子比较携带变化历史；hazard pointer、纪元回收和引用计数主要管理对象何时可以销毁；LL/SC 则依赖平台提供的条件存储语义。它们各自处理的问题和边界不同，不能互相简单替代。无锁设计时，应先明确状态不变量与对象生命周期，再选择机制；如果复杂度超过收益，使用锁往往是更可靠的工程决策。