如何在Golang中使用RWMutex？

Question

type Stat struct {
    counters     map[string]*int64
    countersLock sync.RWMutex
    averages     map[string]*int64
    averagesLock sync.RWMutex
}

它在下面称为

func (s *Stat) Count(name string) {
    s.countersLock.RLock()
    counter := s.counters[name]
    s.countersLock.RUnlock()
    if counter != nil {
        atomic.AddInt64(counter, int64(1))
        return
    }
}

我的理解是我们首先锁定接收器s(这是一个类型Stat),然后如果计数器存在则我们添加它.

问题:

Q1:为什么我们需要锁定它？是什么RWMutex,甚至什么意思？

Q2:s.countersLock.RLock()- 这会锁定整个接收器还是只锁定Stat类型的计数器字段？

问题3:s.countersLock.RLock()- 这会锁定平均值字段吗？

Q4:我们为什么要使用RWMutex？我认为channel是在Golang中处理并发的首选方式？

Q5:这是什么atomic.AddInt64.在这种情况下,为什么我们需要原子？

问题6:为什么我们会在添加之前解锁？

Answer 1

当多个线程*需要改变相同的值时,需要一个锁定机制来同步访问.没有它,两个或多个线程*可能同时写入相同的值,导致内存损坏,通常会导致崩溃.

该原子包提供了一个快速简便的方法来同步访问的原始值.对于计数器,它是最快的同步方法.它具有定义明确的用例的方法,例如递增,递减,交换等.

的同步包提供了一种同步访问更复杂的值,诸如地图,切片,阵列或组的值.您将此用于未在原子中定义的用例.

无论哪种情况,只有在写入时才需要锁定.多个线程*可以在没有锁定机制的情况下安全地读取相同的值.

让我们来看看你提供的代码.

type Stat struct {
    counters     map[string]*int64
    countersLock sync.RWMutex
    averages     map[string]*int64
    averagesLock sync.RWMutex
}

func (s *Stat) Count(name string) {
    s.countersLock.RLock()
    counter := s.counters[name]
    s.countersLock.RUnlock()
    if counter != nil {
        atomic.AddInt64(counter, int64(1))
        return
    }
}

这里缺少的是地图本身是如何初始化的.到目前为止,这些地图并未发生变异.如果计数器名称是预先确定的,并且以后无法添加,则不需要RWMutex.该代码可能如下所示:

type Stat struct {
    counters map[string]*int64
}

func InitStat(names... string) Stat {
    counters := make(map[string]*int64)
    for _, name := range names {
        counter := int64(0)
        counters[name] = &counter
    }
    return Stat{counters}
}

func (s *Stat) Count(name string) int64 {
    counter := s.counters[name]
    if counter == nil {
        return -1 // (int64, error) instead?
    }
    return atomic.AddInt64(counter, 1)
}

(注意:我删除了平均值,因为它没有在原始示例中使用.)

现在,假设您不希望您的计数器被预先确定.在这种情况下,您需要一个互斥锁来同步访问.

让我们只用一个Mutex来试试吧.这很简单,因为一次只能有一个线程*保持锁定.如果第二个线程*在第一个线程*尝试锁定之前在第一次释放它们时使用解锁,它会等待(或阻止)**直到那时.

type Stat struct {
    counters map[string]*int64
    mutex    sync.Mutex
}

func InitStat() Stat {
    return Stat{counters: make(map[string]*int64)}
}

func (s *Stat) Count(name string) int64 {
    s.mutex.Lock()
    counter := s.counters[name]
    if counter == nil {
        value := int64(0)
        counter = &value
        s.counters[name] = counter
    }
    s.mutex.Unlock()
    return atomic.AddInt64(counter, 1)
}

上面的代码可以正常工作.但是有两个问题.

1. 如果Lock()和Unlock()之间出现混乱,即使您要从恐慌中恢复,互斥锁也会永久锁定.这段代码可能不会引起恐慌,但总的来说,假设它可能是更好的做法.
2. 取出计数器时会进行独占锁定.一次只能有一个线程*从计数器读取.

问题#1很容易解决.使用延期:

func (s *Stat) Count(name string) int64 {
    s.mutex.Lock()
    defer s.mutex.Unlock()
    counter := s.counters[name]
    if counter == nil {
        value := int64(0)
        counter = &value
        s.counters[name] = counter
    }
    return atomic.AddInt64(counter, 1)
}

这确保始终调用Unlock().如果由于某种原因你有多个返回,你只需要在函数的头部指定一次Unlock().

问题#2可以用RWMutex解决.它是如何工作的,为什么它有用？

RWMutex是Mutex的扩展,并添加了两个方法:RLock和RUnlock.关于RWMutex,有几点值得注意:

• RLock是一个共享读锁.当用它锁定时,其他线程*也可以用RLock自己锁定.这意味着多个线程*可以同时读取.这是半独家的.

• 如果读取锁定的互斥锁,则阻止对Lock的调用**.如果一个或多个读者持有锁,则无法写入.

• 如果互斥锁被写入锁定(使用Lock),则RLock将阻止**.

考虑它的一个好方法是RWMutex是一个带读卡器计数器的Mutex.当RUnlock递减时,RLock递增计数器.只要该计数器> 0,对Lock的调用就会阻塞.

您可能会想:如果我的应用程序读得很重,那是否意味着作者可以无限期地被阻止？不.RWMutex还有一个有用的属性:

• 如果读取器计数器大于0并且调用了Lock,则对RLock的未来调用也将阻塞,直到现有读取器已释放其锁定,编写器已获得锁定并稍后释放它.

可以把它想象成杂货店登记册上面的灯,上面写着收银员是否开放.排队的人会留在那里,他们会得到帮助,但新人不能排队.一旦最后剩下的客户得到帮助,收银员就会休息,并且该登记册要么一直关闭,直到他们回来或者他们被另一个收银员替换.

让我们用RWMutex修改前面的例子:

type Stat struct {
    counters map[string]*int64
    mutex    sync.RWMutex
}

func InitStat() Stat {
    return Stat{counters: make(map[string]*int64)}
}

func (s *Stat) Count(name string) int64 {
    var counter *int64
    if counter = getCounter(name); counter == nil {
        counter = initCounter(name);
    }
    return atomic.AddInt64(counter, 1)
}

func (s *Stat) getCounter(name string) *int64 {
    s.mutex.RLock()
    defer s.mutex.RUnlock()
    return s.counters[name]
}

func (s *Stat) initCounter(name string) *int64 {
    s.mutex.Lock()
    defer s.mutex.Unlock()
    counter := s.counters[name]
    if counter == nil {
        value := int64(0)
        counter = &value
        s.counters[name] = counter    
    }
    return counter
}

通过上面的代码,我将逻辑分离为getCounter和initCounter函数:

• 保持代码易于理解.RLock()和Lock()在同一个函数中很难.
• 使用延迟时尽早释放锁.

与Mutex示例不同,上面的代码允许您同时增加不同的计数器.

我想指出的另一件事是上面的所有例子,地图map[string]*int64包含指向计数器的指针,而不是计数器本身.如果您要将计数器存储在地图中map[string]int64,则需要使用没有原子的Mutex.该代码看起来像这样:

type Stat struct {
    counters map[string]int64
    mutex    sync.Mutex
}

func InitStat() Stat {
    return Stat{counters: make(map[string]int64)}
}

func (s *Stat) Count(name string) int64 {
    s.mutex.Lock()
    defer s.mutex.Unlock()
    s.counters[name]++
    return s.counters[name]
}

您可能希望这样做以减少垃圾收集 - 但只有当您有数千个计数器时才会这么做 - 即使这样,计数器本身也不会占用大量空间(与字节缓冲区相比).

*当我说线程我的意思是常规.其他语言的线程是一种同时运行一组或多组代码的机制.创建和拆除线程的成本很高.go-routine建立在线程之上,但重用它们.当一个go例程休眠时,底层线程可以被另一个例程使用.当一个go-routine唤醒时,它可能在另一个线程上.Go在幕后处理所有这些. - 但是对于所有意图和目的,当涉及到内存访问时,你会像线程一样处理一个go-routine.但是,在执行线程时使用go-routines时不必保守.

**当复例程由阻塞Lock,RLock,通道,或睡眠,可能被重新使用的底层线程.该例程没有使用cpu - 将其视为排队等候.像其他语言一样,无限循环for {}就会阻止cpu和常规忙碌 - 想到这就像在一个圆圈中跑来跑去 - 你会头晕,呕吐,周围的人不会很开心.

Answer 2

问题:

Q1:为什么我们需要锁定它？是什么RWMutex,甚至什么意思？

RW代表读/写.CF doc:http://golang.org/pkg/sync/#RWMutex.

我们需要锁定它以防止其他例程/线程在我们处理它时更改值.

Q2:s.countersLock.RLock()- 这会锁定整个接收器还是只锁定Stat类型的计数器字段？

作为互斥锁,只有在调用RLock()函数时才会发生锁定.如果任何其他goroutine已经调用了WLock(),那么它会阻塞.你可以RLock()在同一个goroutine中调用任意数量,它不会锁定.

因此它不会锁定任何其他字段,甚至不会s.counters.在您的示例中,您锁定地图查找以查找正确的计数器.

问题3:s.countersLock.RLock()- 这会锁定平均值字段吗？

不,如第二季度所述,RLock只有自己锁定.

Q4:我们为什么要使用RWMutex？我认为channel是在Golang中处理并发的首选方式？

频道是非常有用的,但有时它是不够的,有时它没有意义.

在这里,当您锁定地图访问时,互斥锁是有意义的.使用chan,你必须有1缓冲的chan,之前发送和之后接收.不是很直观.

Q5:这是什么atomic.AddInt64.在这种情况下,为什么我们需要原子？

此函数将以原子方式递增给定变量.在你的情况下,你有一个竞争条件:counter是一个指针,并且在释放锁之后和调用之前可以销毁实际变量atomic.AddInt64.如果你不熟悉这类东西,我建议你坚持使用互斥锁,并在锁定/解锁之间进行所需的所有处理.

问题6:为什么我们会在添加之前解锁？

你不应该.

我不知道你想做什么,但这是一个(简单)的例子:https://play.golang.org/p/cVFPB-05dw

Answer 3

让我们将其与常规的进行比较sync.Mutex，常规的在给定时间只有一个消费者可以持有锁。打个有趣的比方：想象一杯美味的草莓奶昔，需要由一群朋友（消费者）分享。

朋友们想分享奶昔，并决定使用一根专用吸管（锁），因此在给定时间只有一个朋友可以用吸管喝。朋友打电话m.Lock()表示他们想喝酒。如果没有人在喝酒，他们就继续喝，但其他人已经在使用吸管，他们必须等待（阻塞），直到前一个朋友喝完并呼叫m.Unlock()他们。

\\  |  |
 \\ |__|

m.Lock()
m.Unlock()

让我们进入sync.RWMutex（读写互斥体），其中任意数量的读取器都可以持有锁，或者单个写入者可以持有锁。

以草莓奶昔为例，朋友们决定与许多“读者”吸管和一根专有的“作家”吸管分享奶昔。朋友打电话m.RLock()表示他们想用一根“读者”吸管喝酒，并且可以同时与其他读者一起喝酒。然而，独家“作家”稻草的作品却像以前一样。当有人打电话时m.Lock()，他们表示想单独喝酒。在那一刻，每个人都被封锁，直到所有“读者”吸管都喝完（调用m.RUnlock()）。然后，独家撰稿人开始独自喝酒。任何其他对m.RLock()或 的调用m.Lock()都必须等到拥有专属“作家”吸管的朋友喝完为止（直到他们调用m.Unlock()）。

\\  |  |   //  //  //  //
 \\ |__|  //  //  //  //  ...

m.Lock()         m.RLock()
m.Unlock()       m.RUnlock()

使用术语“读者”和“作者”是因为这是最常见的情况。并发内存读取很好，但写入必须是顺序的。如果一个进程试图读取内存地址，而另一个进程正在写入，则可能会导致内存损坏。