Goroutines
Goroutine是Go中最基本的执行单元。事实上每一个Go程序至少有一个goroutine:主goroutine。当程序启动时,它会自动创建。
事实上goroutine采用了一种fork-join的模型。
sayHello := func() {
fmt.Println("hello")
}
go sayHello()
那我们如何来join goroutine呢?需要引入wait操作:
var wg sync.WaitGroup()
sayHello := func() {
defer wg.Done()
fmt.Println("hello")
}
wg.Add(1)
go sayHello()
wa.Wait()
Channel
读写channel
goroutine是Go语言的基本调度单位,而channels则是它们之间的通信机制。操作符<-用来指定管道的方向,发送或接收。如果未指定方向,则为双向管道。
// 创建一个双向channel
ch := make(chan interface{})
interface{}表示chan可以为任何类型
channel有发送和接受两个主要操作。发送和接收两个操作都使用<-运算符。在发送语句中,channel放<-运算符左边。在接收语句中,channel放<-运算符右边。一个不使用接收结果的接收操作也是合法的。
// 发送操作
ch <- x
// 接收操作
x = <-ch
// 忽略接收到的值,合法
<-ch
我们不能弄错channel的方向:
writeStream := make(chan<- interface{})
readStream := make(<-chan interface{})
<-writeStream
readStream <- struct{}{}
关闭channel
Channel支持close操作,用于关闭channel,后面对该channel的任何发送操作都将导致panic异常。对一个已经被close过的channel进行接收操作依然可以接受到之前已经成功发送的数据;如果channel中已经没有数据的话将产生一个零值的数据。
从已经关闭的channel中读:
intStream := make(chan int)
close(intStream)
integer, ok := <- intStream
fmt.Pritf("(%v): %v", ok, integer)
// (false): 0
上面例子中通过返回值ok来判断channel是否关闭,我们还可以通过range这种更优雅的方式来处理已经关闭的channel:
intStream := make(chan int)
go func() {
defer close(intStream)
for i:=1; i<=5; i++{
intStream <- i
}
}()
for integer := range intStream {
fmt.Printf("%v ", integer)
}
// 1 2 3 4 5
带缓冲(buffered)的channel
创建了一个可以持有三个字符串元素的带缓冲Channel:
ch = make(chan string, 3)
我们可以在无阻塞的情况下连续向新创建的channel发送三个值:
ch <- "A"
ch <- "B"
ch <- "C"
此刻,channel的内部缓冲队列将是满的,如果有第四个发送操作将发生阻塞。
如果我们接收一个值:
fmt.Println(<-ch) // "A"
那么channel的缓冲队列将不是满的也不是空的,因此对该channel执行的发送或接收操作都不会发生阻塞。通过这种方式,channel的缓冲队列缓冲解耦了接收和发送的goroutine。
带缓冲的信道可被用作信号量,例如限制吞吐量。在此例中,进入的请求会被传递给 handle,它从信道中接收值,处理请求后将值发回该信道中,以便让该 “信号量” 准备迎接下一次请求。信道缓冲区的容量决定了同时调用 process 的数量上限。
var sem = make(chan int, MaxOutstanding)
func handle(r *Request) {
sem <- 1 // 等待活动队列清空。
process(r) // 可能需要很长时间。
<-sem // 完成;使下一个请求可以运行。
}
func Serve(queue chan *Request) {
for {
req := <-queue
go handle(req) // 无需等待 handle 结束。
}
}
然而,它却有个设计问题:尽管只有 MaxOutstanding 个 goroutine 能同时运行,但 Serve 还是为每个进入的请求都创建了新的 goroutine。其结果就是,若请求来得很快, 该程序就会无限地消耗资源。为了弥补这种不足,我们可以通过修改 Serve 来限制创建 Go 程,这是个明显的解决方案,但要当心我们修复后出现的 Bug。
func Serve(queue chan *Request) {
for req := range queue {
sem <- 1
go func() {
process(req) // 这儿有 Bug,解释见下。
<-sem
}()
}
}
Bug 出现在 Go 的 for 循环中,该循环变量在每次迭代时会被重用,因此 req 变量会在所有的 goroutine 间共享,这不是我们想要的。我们需要确保 req 对于每个 goroutine 来说都是唯一的。有一种方法能够做到,就是将 req 的值作为实参传入到该 goroutine 的闭包中:
func Serve(queue chan *Request) {
for req := range queue {
sem <- 1
go func(req *Request) {
process(req)
<-sem
}(req)
}
}
另一种解决方案就是以相同的名字创建新的变量,如例中所示:
func Serve(queue chan *Request) {
for req := range queue {
req := req // 为该 Go 程创建 req 的新实例。
sem <- 1
go func() {
process(req)
<-sem
}()
}
}
下面再看一个Go语言圣经的例子。它并发地向三个镜像站点发出请求,三个镜像站点分散在不同的地理位置。它们分别将收到的响应发送到带缓冲channel,最后接收者只接收第一个收到的响应,也就是最快的那个响应。因此mirroredQuery函数可能在另外两个响应慢的镜像站点响应之前就返回了结果。
func mirroredQuery() string {
responses := make(chan string, 3)
go func() { responses <- request("asia.gopl.io") }()
go func() { responses <- request("europe.gopl.io") }()
go func() { responses <- request("americas.gopl.io") }()
// 仅仅返回最快的那个response
return <-responses
}
func request(hostname string) (response string) { /* ... */ }
如果我们使用了无缓冲的channel,那么两个慢的goroutines将会因为没有人接收而被永远卡住。这种情况,称为goroutines泄漏,这将是一个BUG。和垃圾变量不同,泄漏的goroutines并不会被自动回收,因此确保每个不再需要的goroutine能正常退出是重要的。
Channels of channels
Go 最重要的特性就是信道是first-class value,它可以被分配并像其它值到处传递。 这种特性通常被用来实现安全、并行的多路分解。
我们可以利用这个特性来实现一个简单的RPC。
以下为 Request 类型的大概定义。
type Request struct {
args []int
f func([]int) int
resultChan chan int
}
客户端提供了一个函数及其实参,此外在请求对象中还有个接收应答的信道。
func sum(a []int) (s int) {
for _, v := range a {
s += v
}
return
}
request := &Request{[]int{3, 4, 5}, sum, make(chan int)}
// 发送请求
clientRequests <- request
// 等待回应
fmt.Printf("answer: %d\n", <-request.resultChan)
服务端的handler函数:
func handle(queue chan *Request) {
for req := range queue {
req.resultChan <- req.f(req.args)
}
}
Channels pipeline
Channels也可以用于将多个goroutine连接在一起,一个Channel的输出作为下一个Channel的输入。这种串联的Channels就是所谓的管道(pipeline)。下面的程序用两个channels将三个goroutine串联起来:
第一个goroutine是一个计数器,用于生成0、1、2、……形式的整数序列,然后通过channel将该整数序列发送给第二个goroutine;第二个goroutine是一个求平方的程序,对收到的每个整数求平方,然后将平方后的结果通过第二个channel发送给第三个goroutine;第三个goroutine是一个打印程序,打印收到的每个整数。
func counter(out chan<- int) {
for x := 0; x < 100; x++ {
out <- x
}
close(out)
}
func squarer(out chan<- int, in <-chan int) {
for v := range in {
out <- v * v
}
close(out)
}
func printer(in <-chan int) {
for v := range in {
fmt.Println(v)
}
}
func main() {
naturals := make(chan int)
squares := make(chan int)
go counter(naturals)
go squarer(squares, naturals)
printer(squares)
}
Select多路复用
select用于从一组可能的通讯中选择一个进一步处理。如果任意一个通讯都可以进一步处理,则从中随机选择一个,执行对应的语句。否则,如果又没有默认分支(default case),select语句则会阻塞,直到其中一个通讯完成。
select {
case <-ch1:
// ...
case x := <-ch2:
// ...use x...
case ch3 <- y:
// ...
default:
// ...
}
如何使用select语句为一个操作设置一个时间限制。代码会输出变量news的值或者超时消息,具体依赖于两个接收语句哪个先执行:
select {
case news := <-NewsAgency:
fmt.Println(news)
case <-time.After(time.Minute):
fmt.Println("Time out: no news in one minute.")
}
下面的select语句会在abort channel中有值时,从其中接收值;无值时什么都不做。这是一个非阻塞的接收操作;反复地做这样的操作叫做“轮询channel”。
select {
case <-abort:
fmt.Printf("Launch aborted!\n")
return
default:
// do nothing
}
参考资料。
有疑问加站长微信联系(非本文作者)