进一步认识Golang中的并发

文章目录

1. sync.WaitGroup
2. channel
3. 使用channel控制goroutine数量

go提供了sync包和channel机制来解决协程间的同步与通信。channel的用法非常灵活，使用的方式多种多样，而且官网的Effective Go中给出了channel的一种并发以外的方式。我们先来介绍sync包提供的调度支持吧。

sync.WaitGroup

WaitGroup用于等待一组线程的结束。父线程调用Add方法来设定应等待的线程的数量。每个被等待的线程在结束时应调用Done方法。同时，主线程里可以调用Wait方法阻塞至所有线程结束。具体用法参考如下代码：

// 代码粘上就可以跑通
package main

import (
	"fmt"
	"sync"
)

var wg sync.WaitGroup

func main() {
	for i := 0; i < 10; i++ {
		wg.Add(1) //每创建一个goroutine，就把任务队列中任务的数量+1
		go func(shownum int) {
			defer wg.Done() //任务完成，将任务队列中的任务数量-1，其实.Done就是.Add(-1)
			fmt.Println(shownum)
		}(i)
	}
	wg.Wait() //.Wait()这里会发生阻塞，直到队列中所有的任务结束就会解除阻塞
}

我们可以利用sync.WaitGroup来满足这样的情况：

某个地方需要创建多个goroutine，并且一定要等它们都执行完毕后再继续执行接下来的操作。

是的，WaitGroup最大的优点就是.Wait()可以阻塞到队列中的任务都完毕后才解除阻塞。

channel

channel是一种golang内置的类型，英语的直译为”通道”，其实，它真的就是一根管道，而且是一个先进先出的数据结构。

我们能对channel进行的操作只有4种：

创建chennel (通过make()函数)
放入数据 (通过 channel <- data 操作)
取出数据 (通过 <-channel 操作)
关闭channel (通过close()函数)

但是channel有一些非常给力的性质需要你牢记，请一定要记住并理解好它们：

(1) channel是一种阻塞管道，是自动阻塞的。 意思就是，如果管道满了，一个对channel放入数据的操作就会阻塞，直到有某个routine从channel中取出数据，这个放入数据的操作才会执行。相反同理，如果管道是空的，一个从channel取出数据的操作就会阻塞，直到某个routine向这个channel中放入数据，这个取出数据的操作才会执行。这事channel最重要的一个性质，没有之一。

package main

func main() {
	ch := make(chan int, 3)
	ch <- 1
	ch <- 1
	ch <- 1
	ch <- 1 //这一行操作就会发生阻塞，因为前三行的放入数据的操作已经把channel填满了
}

package main

func main() {
	ch := make(chan int, 3)
	<-ch //这一行会发生阻塞，因为channel才刚创建，是空的，没有东西可以取出
}

(2)channel分为有缓冲的channel和无缓冲的channel。两种channel的创建方法如下：

1 2	ch := make(chan int) //无缓冲的channel，同等于make(chan int, 0) ch := make(chan int, 5) //一个缓冲区大小为5的channel

操作一个channel时一定要注意其是否带有缓冲，因为有些操作会触发channel的阻塞导致死锁。下面就来解释这些需要注意的情景。
首先来看一个一个例子，这个例子是两段只有主函数不同的代码：

package main

import "fmt"

func Afuntion(ch chan int) {
	fmt.Println("finish")
	<-ch
}

func main() {
	ch := make(chan int) //无缓冲的channel
	go Afuntion(ch)
	ch <- 1
	
	// 输出结果：
	// finish
}

package main

import "fmt"

func Afuntion(ch chan int) {
	fmt.Println("finish")
	<-ch
}

func main() {
	ch := make(chan int) //无缓冲的channel
	//只是把这两行的代码顺序对调一下
	ch <- 1
	go Afuntion(ch)

	// 输出结果：
	// 死锁，无结果
}

前一段代码最终会输出”finish”并正常结束，但是后一段代码会发生死锁。为什么会出现这种现象呢，咱们把上面两段代码的逻辑跑一下。

第一段代码：

创建了一个无缓冲channel
启动了一个goroutine，这个routine中对channel执行取出操作，但是因为这时候channel为空，所以这个取出操作发生阻塞，但是主routine可没有发生阻塞，它还在继续运行呢
主goroutine这时候继续执行下一行，往channel中放入了一个数据
这时阻塞的那个routine检测到了channel中存在数据了，所以接触阻塞，从channel中取出数据，程序就此完毕

第二段代码：

创建了一个无缓冲的channel
主routine要向channel中放入一个数据，但是因为channel没有缓冲，相当于channel一直都是满的，所以这里会发生阻塞。可是下面的那个goroutine还没有创建呢，主routine在这里一阻塞，整个程序就只能这么一直阻塞下去了，然后。。。然后就没有然后了。。死锁！

从这里可以看出，对于无缓冲的channel，放入操作和取出操作不能再同一个routine中，而且应该是先确保有某个routine对它执行取出操作，然后才能在另一个routine中执行放入操作。

对于带缓冲的channel，就没那么多讲究了，因为有缓冲空间，所以只要缓冲区不满，放入操作就不会阻塞，同样，只要缓冲区不空，取出操作就不会阻塞。而且，带有缓冲的channel的放入和取出可以用在同一个routine中。

但是，并不是说有了缓冲就可以随意使用channel的放入和取出了，我们一定要注意放入和取出的速率问题。下面我们就举个例子来说明这种问题：
我们经常会用利用channel自动阻塞的性质来控制当前运行的goroutine的总数量，如下：

package main

import (
	"fmt"
)

func Afunction(ch chan int) {
	fmt.Println("finish")
	<-ch //goroutine执行完了就从channel取出一个数据
}

func main() {
	ch := make(chan int, 10)
	for i := 0; i < 1000; i++ {
		//每当创建goroutine的时候就向channel中放入一个数据，如果里面已经有10个数据了，就会
		//阻塞，由此我们将同时运行的goroutine的总数控制在<=10个的范围内
		ch <- 1
		go Afunction(ch)
	}
	// 这里只是示范个例子，当然，接下来应该有些更加周密的同步操作
}

上面这种channel的使用方式几乎经常会用到，但是再看一下接下来这段代码，它和上面这种使用channel的方式几乎一样，但是它会造成问题：

package main

func Afunction(ch chan int) {
	ch <- 1
	ch <- 1
	ch <- 1
	ch <- 1
	ch <- 1

	<-ch
}

func main() {
	//主routine的操作同上面那段代码
	ch := make(chan int, 10)
	for i := 0; i < 100; i++ {
		ch <- 1
		go Afunction(ch)
	}

	// 这段代码运行的结果为死锁
}

上面这段运行和之前那一段基本上原理是一样的，但是运行后却会发生死锁。为什么呢？其实总结起来就一句话，”放得太快，取得太慢了”。
按理说，我们应该在我们主routine中创建子goroutine并每次向channel中放入数据，而子goroutine负责从channel中取出数据。但是我们的这段代码在创建了子goroutine后，每个routine会向channel中放入5个数据。这样，每向channel中放入6个数据才会执行一次取出操作，这样一来就可能会有某一时刻，channel已经满了，但是所有的routine都在执行放入操作(因为它们当前执行放入操作的概率是执行取出操作的6倍)，这样一来，所有的routine都阻塞了，从而导致死锁。

在使用带缓冲的channel时一定要注意放入与取出的速率问题。

(3)关闭后的channel可以取数据，但是不能放数据。而且，channel在执行了close()后并没有真的关闭，channel中的数据全部取走之后才会真正关闭。

package main

func main() {
	ch := make(chan int, 5)
	ch <- 1
	ch <- 1
	close(ch)
	ch <- 1 //不能对关闭的channel执行放入操作
        
        // 会触发panic
}

package main

func main() {
	ch := make(chan int, 5)
	ch <- 1
	ch <- 1
	close(ch)
	<-ch //只要channel还有数据，就可能执行取出操作

        //正常结束
}

package main

import "fmt"

func main() {
	ch := make(chan int, 5)
	ch <- 1
	ch <- 1
	ch <- 1
	ch <- 1
	close(ch)  //如果执行了close()就立即关闭channel的话，下面的循环就不会有任何输出了
	for {
		data, ok := <-ch
		if !ok {
			break
		}
		fmt.Println(data)
	}
	
	// 输出：
	// 1
	// 1
	// 1
	// 1
	// 
	// 调用了close()后，只有channel为空时，channel才会真的关闭
}

使用channel控制goroutine数量

channel的性质到这里就介绍完了，但是看上去，channel的使用似乎比WaitGroup要注意更多的细节，那么有什么理由一定要用channel来实现同步呢？channel相比WaitGroup有一个很大的优点，就是channel不仅可以实现协程的同步，而且可以控制当前正在运行的goroutine的总数。

下面就介绍几种利用channel控制goroutine数量的方法：

一.如果任务数量是固定的：

package main

func Afunction(ch chan int) {
	ch <- 1
}

func main() {
	var (
		ch        chan int = make(chan int, 20) //可以同时运行的routine数量为20
		dutycount int      = 500
	)
	for i := 0; i < dutycount; i++ {
		go Afunction(ch)
	}

	//知道了任务总量，可以像这样利用固定循环次数的循环检测所有的routine是否工作完毕
	for i := 0; i < dutycount; i++ {
		<-ch
	}
}

二.如果任务的数量不固定

package main

import (
	"fmt"
)

func Afunction(routineControl chan int, feedback chan string) {
	defer func() {
		<-routineControl
		feedback <- "finish"
	}()

	// do some process
	// ...
}

func main() {
	var (
		routineCtl chan int    = make(chan int, 20)
		feedback   chan string = make(chan string, 10000)

		msg      string
		allwork  int
		finished int
	)
	for i := 0; i < 1000; i++ {
		routineCtl <- 1
		allwork++
		go Afunction(routineCtl, feedback)
	}

	for {
		msg = <-feedback
		if msg == "finish" {
			finished++
		}
		if finished == allwork {
			break
		}
	}
}

转自这里

iBlog

Write down what I think.

进一步认识Golang中的并发

sync.WaitGroup

channel

使用channel控制goroutine数量