最近写项目，需要用到信号量等待一些资源完成，但是最多等待N毫秒。
在C语言里，有如下的API来实现带超时的信号量等待：
|
1 2 3 4 5 |
SYNOPSIS #include <pthread.h> int pthread_cond_timedwait(pthread_cond_t *cond, pthread_mutex_t *mutex, const struct timespec *abstime); |
然后在查看golang的document后，发现golang里并没有实现带超时的信号量，官方文档在这里。
原理
我的业务场景是这样的：我有一个缓存字典，当多个用户请求1个不存在的key时，只有1个请求会穿透到后端，而所有用户都要排队等这个请求完成，或者超时返回。
怎么实现呢？其实稍微想一想cond的原理，就能模拟一个带超时的cond出来。
在golang里，要同时实现”挂起等待”和”超时返回”，一般得用select case语法，一个case等待阻塞的资源，一个case等待一个timer，这一点是非常确定的。
原本阻塞的资源应该通过条件变量的机制来实现完成通知，既然这里决定用select case，那么自然想到用channel来代替这个完成通知。
接下来的问题就是，很多请求者并发来获取这个资源，但是资源还没有准备好，所以大家都要排队并挂起，等待资源完成，并且当资源完成后通知大家。
所以，这里很自然要为这个资源做一个队列，每个请求者创建一个chan，并将chan放到队列里，接着select case等待这个chan的通知。而另一端，资源完成后遍历队列，通知每个chan即可。
最后一个问题是，只有第一个请求者才能穿透请求到后端，而后续请求者不应该穿透重复的请求，这可以通过判断缓存里是否有这个key作为判定首次的条件，而标记位init来判断请求者是否应该排队。
我的场景
上面是思路，下面是我的业务场景实现。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 |
func (cache *Cache) Get(key string, keyType int) *string { if keyType == KEY_TYPE_DOMAIN { key = "#" + key } else { key = "=" + key } cache.mutex.Lock() item, existed := cache.dict[key] if !existed { item = &cacheItem{} item.key = &key item.waitQueue = list.New() cache.dict[key] = item } cache.mutex.Unlock() conf := config.GetConfig() lastGet := getCurMs() item.mutex.Lock() item.lastGet = lastGet if item.init { // 已存在并且初始化 defer item.mutex.Unlock() return item.value } // 未初始化，排队等待结果 wait := waitItem{} wait.wait_chan = make(chan *string, 1) item.waitQueue.PushBack(&wait) item.mutex.Unlock() // 新增key, 启动goroutine获取初始值 if !existed { go cache.initCacheItem(item, keyType) } timer := time.NewTimer(time.Duration(conf.Cache_waitTime) * time.Millisecond) var retval *string = nil // 等待初始化完成 select { case retval = <- wait.wait_chan: case <- timer.C: } return retval } |
简述一下整个过程：
- 首先锁字典，如果key不存在，说明我是第一个请求者，我会创建这个key对应的value，只不过init=false表示它正在初始化。最后，释放字典锁。
- 接下来，锁住这个key，判断它已经初始化完成，那么直接返回value。否则，创建一个chan放入waitQueue等待队列。最后，释放key锁。
- 接着，如果当前是第一个请求者，那么会穿透请求到后端（在一个独立的协程里去发起网络调用）。
- 现在，创建一个用于超时的定时器。
- 最后，无论当前是否是key的第一个请求者，还是初始化期间的并发请求者，它们都通过select case超时的等待结果完成。
在initCacheItem函数里，数据已获取成功
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 |
// 一旦标记为init, 后续请求将不再操作waitQueue item.mutex.Lock() item.value = newValue item.init = true item.expire = expire item.mutex.Unlock() // 唤醒所有排队者 waitQueue := item.waitQueue for elem := waitQueue.Front(); elem != nil; elem = waitQueue.Front() { wait := elem.Value.(*waitItem) wait.wait_chan <- newValue waitQueue.Remove(elem) } |
- 首先，锁住key，标记init=true，并赋值value，并释放锁。此后的请求，都可以立即返回，无需排队。
- 之后，因为init=true已被标记，此刻再也有没有请求会修改waitQueue，所以无需加锁，直接遍历队列，通知其中的每个chan。
最后
这样就实现了带超时的条件变量效果，实际上我的场景是一个broadcast的cond例子，大家可以参照思路实现自己想要的效果，活学活用。
