《The.Go.Programming.Language.2015.11》之Goroutines

并行处理图片并等待全部图片处理结束

func makeThumbnails6(filenames <-chan string) int64 {
 sizes := make(chan int64)
 var wg sync.WaitGroup // number of working goroutines
 for f := range filenames {
 wg.Add(1)
 // worker
 go func(f string) {
 defer wg.Done()
 thumb, err := thumbnail.ImageFile(f)
 if err != nil {
 log.Println(err)
 return
 }
 info, _ := os.Stat(thumb) // OK to ignore error
 sizes <- info.Size()
 }(f)
 }
 // closer
 go func() {
 wg.Wait()
 close(sizes)
 }()
 var total int64
 for size := range sizes {
 total += size
 }
 return total
}

该代码中var wg sync.WaitGroup用于实现等待全部图片处理结束。
函数makeThumbnails6在执行到

for size := range sizes {

时会阻塞,直到

 go func() {
 wg.Wait()
 close(sizes)
 }()

执行到close(sizes)。该goroutine先阻塞在wg.Wait()上,直到所有的图片处理结束,图片处理结束时会调用wg.Done()。

限定并发数量

最简单的是通过带缓存的通道实现
如下,在执行之前调用tokens <- struct{}{}获取一个名额,执行结束后调用<-tokens释放名额。

// tokens is a counting semaphore used to
// enforce a limit of 20 concurrent requests.
var tokens = make(chan struct{}, 20)
func crawl(url string) []string {
 fmt.Println(url)
 tokens <- struct{}{} // acquire a token
 list, err := links.Extract(url)
 <-tokens // release the token
 if err != nil {
 log.Print(err)
 }
 return list
}

复用select

time.Tick创建一个通道,每隔一段时间往通道里写入数据。

func main() {
 fmt.Println("Commencing countdown.")
 tick := time.Tick(1 * time.Second)
 for countdown := 10; countdown > 0; countdown-- {
 fmt.Println(countdown)
 <-tick
 }
 launch()
}

等待多个通道时使用select

select {
case <-ch1:
// ...
case x := <-ch2:
// ...use x...
case ch3 <- y:
// ...
default:
// ...
}

select阻塞直到某个通道可用,如果多个同时可用,就随机选一个。

time.After在经过指定时间后生成一个通道,同时产生一个goroutine往这个通道里写入一个值。

func main() {
 // ...create abort channel...
 fmt.Println("Commencing countdown. Press return to abort.")
 select {
 case <-time.After(10 * time.Second):
 // Do nothing.
 case <-abort:
 fmt.Println("Launch aborted!")
 return
 }
 launch()
}

定时器的常用使用方式为:

ticker := time.NewTicker(1 * time.Second)
<-ticker.C // receive from the ticker's channel
ticker.Stop() // cause the ticker's goroutine to terminate

非阻塞模式

select {
 case <-abort:
 fmt.Printf("Launch aborted!\n")
 return
 default:
 // do nothing
}

channel的0值是nil。在nil的通道上读写会一直阻塞。select中case

互斥锁

var (
 mu
 sync.Mutex // guards balance
 balance int
)
func Deposit(amount int) {
 mu.Lock()
 balance = balance + amount
 mu.Unlock()
}
func Balance() int {
 mu.Lock()
 defer mu.Unlock()
 return balance
}

文中又举了这样一个例子,一个人如果账户余额不足的话就会回退之前的操作,Deposit(-amount)表示购买操作, Balance()判断余额不足:

// NOTE: not atomic!
func Withdraw(amount int) bool {
 Deposit(-amount)
 if Balance() < 0 {
 Deposit(amount)
 return false // insufficient funds
 }
 return true
}

这个代码的问题在于,假如A有100元,想要购买300的车,Deposit(-amount)造成了Balance()为负数,那么如果在Deposit(amount)之前A打算买小于100元的生活用品也无法购买,因为此时Balance()还为负数,还没有恢复。
所以要把购买和恢复作为一个原子操作。

func Withdraw(amount int) bool {
 mu.Lock()
 defer mu.Unlock()
 deposit(-amount)
 if balance < 0 {
 deposit(amount)
 return false // insufficient funds
 }
 return true
}
func Deposit(amount int) {
 mu.Lock()
 defer mu.Unlock()
 deposit(amount)
}
func Balance() int {
 mu.Lock()
 defer mu.Unlock()
 return balance
}
// This function requires that the lock be held.
func deposit(amount int) {
 balance += amount
}

sync.Once

首先我们看以下代码

var icons map[string]image.Image
func loadIcons() {
 icons = map[string]image.Image{
 "spades.png": loadIcon("spades.png"),
 "hearts.png": loadIcon("hearts.png"),
 "diamonds.png": loadIcon("diamonds.png"),
 "clubs.png": loadIcon("clubs.png"),
 }
}
// NOTE: not concurrency-safe!
func Icon(name string) image.Image {
 if icons == nil {
 loadIcons() // one-time initialization
 }
 return icons[name]
}

其中函数Icon是not concurrency-safe的,假如有两个Goroutines A和B,A第一次执行到if icons == nil然后切换到B,B此时判断if icons == nil仍然为真,这样loadIcons就执行了两遍。

解决方法一,使用互斥锁:

var mu sync.Mutex // guards icons
// Concurrency-safe.
func Icon(name string) image.Image {
 mu.Lock()
 defer mu.Unlock()
 if icons == nil {
 loadIcons()
 }
 return icons[name]
}

但是问题在于icons被初始化之后,后续的读取操作不能并行执行。

解决方法二 使用读写锁

var mu sync.RWMutex // guards icons
var icons map[string]image.Image
// Concurrency-safe.
func Icon(name string) image.Image {
 mu.RLock()
 if icons != nil {
 icon := icons[name]
 mu.RUnlock()
 return icon
 }
 mu.RUnlock()
 // acquire an exclusive lock
 mu.Lock()
 if icons == nil { // NOTE: must recheck for nil
 loadIcons()
 }
 icon := icons[name]
 mu.Unlock()
 return icon
}

这样的问题就是耗费太大

解决方法三 使用sync.Once

var loadIconsOnce sync.Once
var icons map[string]image.Image
// Concurrency-safe.
func Icon(name string) image.Image {
 loadIconsOnce.Do(loadIcons)
 return icons[name]
}

有疑问加站长微信联系(非本文作者)

• 请尽量让自己的回复能够对别人有帮助
• 支持 Markdown 格式, **粗体**、~~删除线~~、`单行代码`
• 支持 @ 本站用户;支持表情(输入 : 提示),见 Emoji cheat sheet
• 图片支持拖拽、截图粘贴等方式上传