用于同步实体行为的有界“互斥池”

Question

我有一个功能

type Command struct {
  id Uuid
}

handleCommand(cmd Command) 
{
  entity := lookupEntityInDataBase(cmd.Uuid)
  entity.handleCommand(cmd)

  saveEntityInDatabase(entity)

}

但是这个函数可以并行调用，并且实体被假定为非线程安全，导致实体状态和将保存在数据库中的状态的活跃性。

在此函数的开头和结尾使用简单的互斥锁可以解决这个问题，但会导致过于悲观的同步，因为不同实例的实体（即不同的 uuid）应该被允许并行处理它们的命令。

另一种方法是保留

map[uuid]sync.Mutex

的 Map，如果之前没有遇到 uuid，则创建一个新的互斥体，并以线程安全的方式创建它。然而，这将导致运行时遇到的所有 uuid 的映射可能无限增长。

我想过之后清理互斥体，但是这样做是线程安全的，并且意识到另一个线程可能已经在等待互斥体，这会打开很多蠕虫。

我希望我缺少一个非常简单而优雅的解决方案。

Answer 1

确实没有一个优雅的解决方案。这是使用通道的版本：

var m map[string]chan struct{}
var l sync.Mutex

func handleCommand(cmd Command) {

   for {
       l.Lock()
       ch, ok:=m[cmd.Uuid]
       if !ok {
           ch=make(chan struct{})
           m[uuid]=ch
           defer func() {
                l.Lock()
                delete(m,cmd.Uuid)
                close(ch)
                l.Unlock()
           }()
           l.Unlock()
           break
        }
     l.Unlock()
     <-ch
   }
  entity := lookupEntityInDataBase(cmd.Uuid)
  entity.handleCommand(cmd)

  saveEntityInDatabase(entity)
}

Answer 2

Moby 项目有一个库之类的东西，请参阅 https://github.com/moby/locker

一行描述是，

locker 提供了一种创建更细粒度锁定的机制，以帮助释放更多全局锁来处理其他任务。

Answer 3

我知道这个问题有点老了，但我遇到了一个类似的问题，我想将单个元素锁定在唯一的 ID 空间中。我得出以下中间立场：

type locker struct {
  mtxs []sync.Mutex
}

func newLocker(partitionCount uint64) locker {
    return locker{
        mtxs: make([]sync.Mutex, partitionCount),
    }
}

func (l locker) hash(id uint64) uint64 {
    // knuth magic number for integer hashing
    // the art of computer programming vol 3, section 6.4
    return (id * 2654435761) % uint64(len(l.mtxs))
}

func (l locker) lock(id uint64) {
    l.mtxs[l.hash(id)].Lock()
}

func (l locker) unlock(id uint64) {
    l.mtxs[l.hash(id)].Unlock()
}

这里的主要思想是我们拥有的锁比并行单位多得多。然后，通过“足够好”的哈希函数，我们可以获得这些互斥体的近似随机采样。案例分析应该和哈希图查找类似。

如果我们的散列函数命中切片中的相同元素，我们当然仍然可以遇到“过于悲观”的情况，但是任何半途而废的散列函数都应该使预期的情况明显更好。另一方面，我们已将内存使用固定为

partitionCount

互斥体。我可能会在这里遗漏一些东西，但我希望这会有所帮助。