假如在开发一个http接口，在并发调用传入相同参数时，需要加锁，一个一个执行。在并发调用传入不同参数时，需要并发处理，不能加锁，这时需要一个基于key的加锁。
于是我开发了一个`Multilocker`类，用法为：使用`l := NewMultilocker()`获取`Multilocker`对象，使用mu := l.Get("key")获取sync.mutex，然后使用这个mutex加锁解锁，最后使用`l.Put("key")`归还这个sync.mutex
如果同一时间并发调用`Get`传入的为相同的key，则获取的是同一个sync.Mutex，如果同一时间并发调用`Get`传入的为不同的key，则获取的是不同的sync.Mutex
multilock.go
```go
package xsync
import (
"sync"
)
type mutexItem struct {
mu sync.Mutex
counter int64
}
// mutexPool is not concurrent safe
type mutexPool struct {
pool []*mutexItem
}
func (p *mutexPool) Put(mi *mutexItem) {
p.pool = append(p.pool, mi)
}
func (p *mutexPool) Get() *mutexItem {
if len(p.pool) == 0 {
return &mutexItem{}
}
i := len(p.pool) - 1
mi := p.pool[i]
p.pool = p.pool[:i]
return mi
}
type messageGet struct {
key string
c chan *sync.Mutex
}
type messagePut string
// Multilocker is key based multiple locker
type Multilocker struct {
gets chan messageGet
puts chan messagePut
inUse map[string]*mutexItem
done chan struct{}
mp mutexPool
cp sync.Pool
}
// Get get key related sync.Mutex
func (l *Multilocker) Get(key string) *sync.Mutex {
msg := messageGet{
key: key,
// c: l.cp.Get().(chan *sync.Mutex),
c: make(chan *sync.Mutex, 1),
}
l.gets <- msg
mu := <-msg.c
// l.cp.Put(msg.c)
return mu
}
// Put release key related mutexItem
func (l *Multilocker) Put(key string) {
l.puts <- messagePut(key)
}
// Close will stop schedule
func (l *Multilocker) Close() {
close(l.done)
}
func (l *Multilocker) handleGet(msg *messageGet) {
key := msg.key
c := msg.c
mi, ok := l.inUse[key]
if !ok {
mi = l.mp.Get()
l.inUse[key] = mi
}
mi.counter++
c <- &mi.mu
}
func (l *Multilocker) handlePut(msg *messagePut) {
key := string(*msg)
mi, ok := l.inUse[key]
if !ok {
panic("should call lock first")
}
mi.counter--
if mi.counter == 0 {
l.mp.Put(mi)
delete(l.inUse, key)
}
}
func (l *Multilocker) schedule() {
loop:
for {
select {
case msg := <-l.gets:
l.handleGet(&msg)
case msg := <-l.puts:
l.handlePut(&msg)
case <-l.done:
break loop
}
}
}
// NewMultilocker return a new Multilocker
func NewMultilocker() *Multilocker {
l := &Multilocker{
gets: make(chan messageGet, 1000),
puts: make(chan messagePut, 1000),
inUse: make(map[string]*mutexItem),
cp: sync.Pool{
New: func() interface{} { return make(chan *sync.Mutex, 1) },
},
}
go l.schedule()
return l
}
```
multilock_test.go
```go
package xsync
import (
"log"
"strconv"
"sync"
"testing"
"time"
)
func TestSingleLock(t *testing.T) {
l := NewMultilocker()
mu := l.Get("3")
mu.Lock()
mu.Unlock()
l.Put("3")
}
func TestMultiple(t *testing.T) {
log.SetFlags(log.LstdFlags | log.Lmicroseconds)
log.Println("start1")
defer log.Println("end")
l := NewMultilocker()
var n sync.WaitGroup
for i := 0; i < 2000; i++ {
n.Add(1)
i := i
key := strconv.Itoa(i % 2)
// key := strconv.Itoa(i)
go func() {
defer n.Done()
mu := l.Get(key)
mu.Lock()
// log.Printf("%d - %s start", i, key)
time.Sleep(time.Millisecond * 10)
// log.Printf("%d - %s end", i, key)
mu.Unlock()
l.Put(key)
}()
}
n.Wait()
}
```
第一次unlock后就释放整个key的锁： 有点类似cache，第一次取的时候缓存没有，需要加锁获取到缓存里，后续直接从缓存获取。可以参考The Go Programming Language的第9章的`9.7. Example: Concurrent Non-Blocking Cache` 的实现。
这个9.7最后也用了一个单独的channel来实现，因为这样性能比较好。如果使用共享变量，老是要加锁解锁的访问共享变量的内容，并且写的代码容易出bug。使用channel实现的话，代码可读性高，易于理解。
毕竟Go比较著名的谚语是：Don't communicate by sharing memory, share memory by communicating.
我这个场景是同一时间相同key，需要先后执行，不能同一时间执行，所以需要counter。
我先看下你的代码
#4
更多评论
找了下我的代码，大概是这样的
func NewUtil() *Util {
return &Util{
locks: &sync.Map{},
}
}
type Util struct {
locks *sync.Map
}
func (u *Util) Locker(key string) (*Locker, bool) {
newlocker := &Locker{
Map: u.locks,
Key: key,
}
v, ok := u.locks.LoadOrStore(key, newlocker)
return v.(*Locker), ok
}
type Locker struct {
sync.RWMutex
Map *sync.Map
Key string
}
func (l *Locker) Unlock() {
l.RWMutex.Unlock()
l.Map.Delete(l.Key)
}
#2
今日阅读排行
一周阅读排行