puzpuzpuz/xsync

[](https://pkg.go.dev/github.com/puzpuzpuz/xsync/v4) [](https://goreportcard.com/report/github.com/puzpuzpuz/xsync/v4) [](https://codecov.io/gh/puzpuzpuz/xsync) # xsync 适用于 Go 的并发数据结构。旨在为标准 `sync` 包中的部分数据结构提供更具扩展性的替代方案，但不仅限于此。 除了直接的库依赖外，`xsync` 数据结构也存在于其他库的代码中，例如 [Otter](https://github.com/maypok86/otter/blob/8c526307556486ea0337280a4211135720bc29cc/internal/hashmap/map.go) 缓存库。 测试覆盖了并发压力测试，遵循[此处](https://puzpuzpuz.dev/testing-concurrent-code-for-fun-and-profit)所描述的方法。 ## 基准测试 基准测试结果可在[此处](BENCHMARKS.md)查看。我要感谢 [@felixge](https://github.com/felixge)，他十分热心地在一台高性能多核机器上运行了基准测试。 此外，还有一个非严谨、不公平的基准测试，对 Java 的 [j.u.c.ConcurrentHashMap](https://docs.oracle.com/en/java/javase/17/docs/api/java.base/java/util/concurrent/ConcurrentHashMap.html) 和 `xsync.Map` 进行了比较，可在[此处](https://puzpuzpuz.dev/concurrent-map-in-go-vs-java-yet-another-meaningless-benchmark)查看。 ## 用法 最新的 xsync 主版本是 v4，因此在导入该库时应使用 `/v4` 后缀： ``` import ( "github.com/puzpuzpuz/xsync/v4" ) ``` 最低要求 Golang 版本为 1.24。 *v4 之前版本的用户请注意*：v3 和 v4 之间的主要变化是移除了非泛型数据结构，并对 `Map` API 进行了一些改进。旧的 `*Of` 类型作为重命名后数据结构的类型别名保留，以简化迁移过程，例如 `MapOf` 是 `Map` 的别名。虽然 API 存在一些破坏性更改，但迁移过程应该非常简单。 ### Counter `Counter` 是一个条带化的 `int64` 计数器，其灵感来源于 Java 标准库中的 `j.u.c.a.LongAdder` 类。 ``` c := xsync.NewCounter() // increment and decrement the counter c.Inc() c.Dec() // read the current value v := c.Value() ``` 在高竞争场景下，与单一通过原子操作更新的 `int64` 计数器相比，它的表现更好。 ### Map `Map` 类似于基于并发哈希表的 map。它遵循 `sync.Map` 的接口，并带有一些有价值的扩展，例如 `Compute` 或 `Size`。 ``` m := xsync.NewMap[string, string]() m.Store("foo", "bar") v, ok := m.Load("foo") s := m.Size() ``` `Map` 使用了 Cache-Line Hash Table (CLHT) 数据结构的修改版本：https://github.com/LPD-EPFL/CLHT CLHT 的核心理念是将哈希表组织成缓存行大小的桶，以便在所有现代 CPU 上，更新操作只需最少的缓存行传输即可完成。此外，`Get` 操作是无阻塞的，且不涉及对共享内存的写入，因此不需要互斥锁或任何其他类型的锁。得益于这种设计，在所有已考量的场景中，`Map` 的性能都优于 `sync.Map`。`Map` 还使用协作式并行 rehashing：这意味着执行写操作的 goroutine 可以参与并发 rehashing，而不是等待其完成。 除了 CLHT 之外，`Map` 还借鉴了 Java 的 `j.u.c.ConcurrentHashMap`（使用不可变的 K/V 对结构代替原子快照）以及 C++ 的 `absl::flat_hash_map`（又称 SwissTable，包含元内存和基于 SWAR 的查找）的思想。 `Map` 使用内置的 Golang 哈希函数，该函数具有抗 DDOS 攻击的保护。它使用 `maphash.Comparable` 作为默认哈希函数。这意味着每个 map 实例都会获得自己的随机种子数，并且哈希函数会使用该种子来计算哈希码。 除了可用于 map 迭代的 `Range` 和 `All` 方法外，还有一个 `ToPlainMap` 工具函数，用于将 `Map` 转换为 Go 内置的 `map`： ``` m := xsync.NewMap[int, int]() m.Store(42, 42) pm := xsync.ToPlainMap(m) ``` 对于批量条件删除，可以使用 `DeleteMatching`。在缓存使用场景中，当需要删除过期条目时，此方法非常实用： ``` m.DeleteMatching(func(key int, value int) (delete, stop bool) { return key%2 == 0, false // delete even keys }) ``` 对于高性能迭代，可以使用 `RangeRelaxed`（以及用于 Go 1.23+ 迭代器的 `AllRelaxed`）。与 `Range` 不同，它是无锁的。但是，它的一致性较弱：如果在迭代期间同一 key 被并发删除并重新插入，则该 key 可能会被访问多次。 ``` m.RangeRelaxed(func(key int, value int) bool { // process entry return true // continue iteration }) ``` ### UMPSCQueue `UMPSCQueue` 是一个无界多生产者单消费者并发队列。这意味着多个 goroutine 可以向队列发布项，但最多只能有一个 goroutine 消费这些项。与有界队列不同，该队列对队列容量没有限制。 ``` q := xsync.NewUMPSCQueue[string]() // producer inserts an item into the queue; doesn't block // safe to invoke from multiple goroutines inserted := q.Enqueue("bar") // consumer obtains an item from the queue // must be called from a single goroutine item := q.Dequeue() // string ``` `UMPSCQueue` 旨在作为 channel 的替代品。但关键是，它具有无限容量。在许多情况下，这是一个非常糟糕的主意，因为这意味着它永远不会产生背压。换句话说，如果没有元素从队列中被消费，它最终会耗尽所有可用内存并导致进程崩溃。然而，在某些情况下这也是期望的行为，这意味着队列会动态分配更多内存来存储临时的突发流量，从而让生产者在消费者追赶进度的同时永远不会阻塞。 底层数据结构表示为由大段组成的单向链表。每个段都是一个 `T` 类型的切片，并带有对应每个索引的 `sync.WaitGroup`。生产者使用一个原子计数器来确定它们在段中写入值的唯一索引，并在写入值后将相应的 wait group 标记为完成。消费者只需跟踪它想要读取的索引，并等待相应的 wait group 完成。这两个操作都不会获取锁，因此在高竞争负载下表现非常出色。 但请注意，由于不获取锁，多个 goroutine 同时从队列中消费是不安全的。消费者之间必须显式进行同步。这使得单消费者的设置永远不会获取锁，从而显著加快了消费速度。 ### SPSCQueue `SPSCQueue` 是一个有界单生产者单消费者并发队列。这意味着最多只能有一个 goroutine 向队列发布项，同时最多只能有一个 goroutine 消费这些项。 ``` q := xsync.NewSPSCQueue[string](1024) // producer inserts an item into the queue // optimistic insertion attempt; doesn't block inserted := q.TryEnqueue("bar") // consumer obtains an item from the queue // optimistic obtain attempt; doesn't block item, ok := q.TryDequeue() // string ``` 该队列基于这篇[文章](https://rigtorp.se/ringbuffer)中的数据结构。其核心思想是通过分别保留生产者和消费者使用的读取和写入索引的缓存副本，来减少 CPU 缓存一致性流量。 请确保实现适当的退避策略，以处理乐观操作尝试失败的情况。最基本的退避方法是调用 `runtime.Gosched()`。 ### MPMCQueue `MPMCQueue` 是一个有界多生产者多消费者并发队列。 ``` // capacity is rounded up to the next power of 2 (1000 -> 1024) q := xsync.NewMPMCQueue[string](1000) // producer optimistically inserts an item into the queue // optimistic insertion attempt; doesn't block inserted := q.TryEnqueue("bar") // consumer obtains an item from the queue // optimistic obtain attempt; doesn't block item, ok := q.TryDequeue() // string ``` 该队列基于 [MPMCQueue](https://github.com/rigtorp/MPMCQueue) C++ 库中的算法，而该库反过来又参考了 D.Vyukov 的 [MPMC queue](https://www.1024cores.net/home/lock-free-algorithms/queues/bounded-mpmc-queue)。根据以下[分类](https://www.1024cores.net/home/lock-free-algorithms/queues)，该队列是基于数组的，溢出时会失败，提供因果 FIFO，并具有阻塞的生产者和消费者。 该算法的核心思想是通过引入“票据”的概念（即两个计数器的值，分别对应生产者/消费者），来实现并发生产者和消费者的并行化。对其中一个计数器的原子递增是队列操作中唯一明显的竞争点。借助对每个队列项基于轮次的读/写访问，操作的其余部分避免了写操作的竞争。 本质上，`MPMCQueue` 是一种专用队列，适用于在大型多核机器上运行的单个队列存在多个并发生产者和消费者的场景。 为了获得最佳性能，您可能希望将队列大小设置得足够大，比如比生产者/消费者的数量大一个数量级，以便在大多数情况下允许生产者和消费者并行执行其队列操作。 除此之外，请确保实现适当的退避策略，以处理乐观操作尝试失败的情况。最基本的退避方法是调用 `runtime.Gosched()`。 ### RBMutex `RBMutex` 是一个偏向读者的读写互斥锁。该锁可以由多个读者或单个写者持有。 ``` mu := xsync.NewRBMutex() // reader lock calls return a token t := mu.RLock() // the token must be later used to unlock the mutex mu.RUnlock(t) // writer locks are the same as in sync.RWMutex mu.Lock() mu.Unlock() ``` `RBMutex` 基于 BRAVO (Biased Locking for Reader-Writer Locks) 算法的修改版本：https://arxiv.org/pdf/1810.01553.pdf 该算法的思想是在现有的读写互斥锁之上构建，并为读者引入一条快速路径。在快速路径上，读者锁的尝试会根据读者身份（在 Golang 的情况下是一个 token）分散到一个内部数组中。这意味着读者不需要像在 `sync.RWMutex` 中那样竞争单个原子计数器，从而在核心数量上获得更好的可扩展性。 因此，在设计上 `RBMutex` 是一种专用的互斥锁，适用于缓存等绝大多数锁由读者获取、且写锁获取尝试不频繁的场景。在这些场景中，在大型多核机器上 `RBMutex` 的性能应该优于 `sync.RWMutex`。 `RBMutex` 在内部扩展了 `sync.RWMutex` 并将其用作“禁用读者偏向”的回退方案，因此适用相同的语义。唯一明显的区别在于从 `RLock`/`RUnlock` 方法返回的读者 token。 除了阻塞方法外，`RBMutex` 还有用于乐观锁的方法： ``` mu := xsync.NewRBMutex() if locked, t := mu.TryRLock(); locked { // critical reader section... mu.RUnlock(t) } if mu.TryLock() { // critical writer section... mu.Unlock() } ``` ## 开源许可 基于 Apache v2 授权。

标签：EVTX分析, Go, Ruby工具, 并发编程, 开发库, 数据结构, 日志审计