《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]
}