📚 课程目录第3章 PN-Counter实现与单调性验证

3. PN-Counter 实现与单调性验证

PN-Counter（Positive-Negative Counter）是 CRDT 中用于支持既增又减的计数器。其核心思想是正负分离设计：将递增与递减操作分别记录在两个独立的 GCounter 中，最终计数器的值为正 GCounter 的值减去负 GCounter 的值。这一章将带领你从零实现 PN-Counter，验证其单调性约束，并完成单元测试编写。

3.1 正负分离设计原理

PN-Counter 由两个 GCounter 组成：

P（Positive Counter）—— 记录所有增量操作，只增不减。
N（Negative Counter）—— 记录所有减量操作，也只增不减（因为减量操作被抽象为正的“减量计数”）。

最终值 value = P.value() - N.value()。由于两个 GCounter 各自满足单调递增，PN-Counter 的合并操作（merge）要求同时对 P 和 N 执行 GCounter 的 merge，从而保证整体数据一致性。

属性	GCounter	PN-Counter
操作类型	仅递增	递增、递减
内部结构	单字典（ replicaId → count ）	两个字典（ P 和 N ）
并发冲突合并	取 max	分别对 P 和 N 取 max
最终值计算	sum(counts)	sum(P) – sum(N)

    关键性质：虽然 PN-Counter 的总值可能上下波动（负数也可能出现），但 P 和 N 各自内部始终是单调递增的。这种“分离单调性”是 CRDT 正确性的基础。

3.2 实现流程与代码

以下 Python 实现演示了完整的 PN-Counter 类，包含 increment、decrement、merge 和 value 方法。假设你已经实现了 GCounter。

class GCounter:
    def __init__(self):
        self.counts = {}  # replicaId -> count

    def increment(self, replica_id, delta=1):
        self.counts[replica_id] = self.counts.get(replica_id, 0) + delta

    def merge(self, other):
        for rid, cnt in other.counts.items():
            self.counts[rid] = max(self.counts.get(rid, 0), cnt)

    def value(self):
        return sum(self.counts.values())

class PNCounter:
    def __init__(self):
        self.p = GCounter()  # 正计数器
        self.n = GCounter()  # 负计数器

    def increment(self, replica_id, delta=1):
        self.p.increment(replica_id, delta)

    def decrement(self, replica_id, delta=1):
        self.n.increment(replica_id, delta)

    def merge(self, other):
        self.p.merge(other.p)
        self.n.merge(other.n)

    def value(self):
        return self.p.value() - self.n.value()

3.3 单调递增约束验证

单调递增是对 P 和 N 内部状态的约束，而非最终值。验证方法：在合并前后，每个副本在 P 和 N 上的计数都只能保持不变或增加。下面给出验证脚本片段。

def test_monotonic_merge():
    c1 = PNCounter()
    c2 = PNCounter()

    c1.increment("A", 3)
    c1.decrement("A", 1)

    c2.increment("B", 2)
    c2.decrement("B", 1)

    # 记录合并前的 P/N 快照
    snap1_p = dict(c1.p.counts)
    snap1_n = dict(c1.n.counts)
    snap2_p = dict(c2.p.counts)
    snap2_n = dict(c2.n.counts)

    c1.merge(c2)

    # 验证单调递增：合并后每个 replicaId 的计数不小于合并前
    for rid, cnt in snap1_p.items():
        assert c1.p.counts.get(rid, 0) >= cnt
    for rid, cnt in snap2_p.items():
        assert c1.p.counts.get(rid, 0) >= cnt
    for rid, cnt in snap1_n.items():
        assert c1.n.counts.get(rid, 0) >= cnt
    for rid, cnt in snap2_n.items():
        assert c1.n.counts.get(rid, 0) >= cnt

    print("单调递增约束通过")

提示：实际生产环境中每个副本需分配唯一 ID（如 UUID），并确保 increment/decrement 时正确传递该 ID。

3.4 单元测试编写

单元测试覆盖四个核心场景：基本加减、并发合并、负值验证、空合并。推荐使用 Python 内置的 unittest 框架。

import unittest

class TestPNCounter(unittest.TestCase):
    def test_basic(self):
        c = PNCounter()
        c.increment("A", 5)
        c.decrement("A", 2)
        self.assertEqual(c.value(), 3)

    def test_concurrent_merge(self):
        c1 = PNCounter()
        c2 = PNCounter()
        c1.increment("A", 10)
        c1.decrement("A", 3)
        c2.increment("B", 7)
        c2.decrement("B", 4)
        c1.merge(c2)
        c2.merge(c1)          # 双向合并
        self.assertEqual(c1.value(), 10)
        self.assertEqual(c2.value(), 10)

    def test_negative_value(self):
        c = PNCounter()
        c.decrement("A", 8)
        self.assertEqual(c.value(), -8)

    def test_merge_empty(self):
        c1 = PNCounter()
        c2 = PNCounter()
        c1.increment("X", 1)
        c1.merge(c2)          # 合并空计数器不应改变值
        self.assertEqual(c1.value(), 1)

if __name__ == "__main__":
    unittest.main()

常见错误：

在 decrement 中直接对 N 执行减法 —— 破坏单调性。正确做法是对 N 执行递增操作。
合并时只合并 P 而遗漏 N —— 导致丢失减量记录。
未处理不同副本间的并发操作 —— PN-Counter 依靠 GCounter 的 max 合并语义自动解决。

3.5 结构总览图

下面的 SVG 展示了 PN-Counter 的内部组成及合并流程。

3.6 性能与稳定性优化建议

存储空间：每个副本的 P 和 N 词典会随时间增长。可引入定期 GC 清理：对于已经合并到所有节点的副本 ID，可安全删除其旧记录（需配合状态向量或版本向量）。
批量合并：当网络波动导致大量待合并数据时，建议合并前压缩相同副本 ID 的计数（取 max 即可），减少字典遍历开销。
整数溢出：防止 P 或 N 的内部计数过大，使用 64 位整数（Python 自动大整数无此问题，但其他语言需注意）。