1. 背景
go代碼中要實現異步很簡單,go funcName()。
但是進程需要控制協程數量在合理范圍內,對應大批量任務可以使用“協程池 + 無鎖隊列”實現。
2. golang無鎖隊列實現思路
- Channel是Go中的一個核心類型,你可以把它看成一個管道,通過它並發核心單元就可以發送或者接收數據進行通訊(communication)。無鎖隊列使用帶buff的chan存儲數據。
- interface{} (類似c++的void*, java的Object)可以與任意類型互轉。無鎖隊列使用interface{}作為數據存儲類型。
- select可以處理多個信號, 可以用來解決channel阻塞問題。
3. 代碼實現
package main
import (
"fmt"
"time"
)
type DataContainer struct {
Queue chan interface{}
}
func NewDataContainer(max_queue_len int) (dc *DataContainer){
dc = &DataContainer{}
dc.Queue = make(chan interface{}, max_queue_len)
return dc
}
//非阻塞push
func (dc *DataContainer) Push(data interface{}, waittime time.Duration) bool{
click := time.After(waittime)
select {
case dc.Queue <- data:
return true
case <- click:
return false
}
}
//非阻塞pop
func (dc *DataContainer) Pop(waittime time.Duration) (data interface{}){
click := time.After(waittime)
select {
case data =<-dc.Queue:
return data
case <- click:
return nil
}
}
//test
var MAX_WAIT_TIME = 10 *time.Millisecond
func main(){
type dataItem struct {
name string
age int
}
datacotainer := NewDataContainer(2)
//add
fmt.Printf("res=%v\n", datacotainer.Push(&dataItem{"zhangsan",25}, MAX_WAIT_TIME))
fmt.Printf("res=%v\n", datacotainer.Push(&dataItem{"lisi",30}, MAX_WAIT_TIME))
fmt.Printf("res=%v\n", datacotainer.Push(&dataItem{"wangwu",28}, MAX_WAIT_TIME))
//get
var item interface{}
item = datacotainer.Pop(MAX_WAIT_TIME)
if item != nil{
if tmp,ok := item.(*dataItem); ok{ //interface轉為具體類型
fmt.Printf("item name:%v, age:%v\n", tmp.name, tmp.age)
}
}
}