Go语言学习：Channel

本文转载于http://hustcat.github.io/channel/

“网络，并发”是Go语言的两大feature。Go语言号称“互联网的C语言”，与使用传统的C语言相比，写一个Server所使用的代码更少，也更简单。写一个Server除了网络，另外就是并发，相对python等其它语言，Go对并发支持使得它有更好的性能。

Goroutine和channel是Go在“并发”方面两个核心feature。

Channel是goroutine之间进行通信的一种方式，它与Unix中的管道类似。

Channel声明：

ChannelType = ( "chan" | "chan" "<-" | "<-" "chan" ) ElementType .

例如：

var ch chan int
var ch1 chan<- int  //ch1只能写
var ch2 <-chan int  //ch2只能读

channel是类型相关的，也就是一个channel只能传递一种类型。例如，上面的ch只能传递int。

在go语言中，有4种引用类型：slice，map，channel，interface。

Slice，map，channel一般都通过make进行初始化：

ci := make(chan int)            // unbuffered channel of integers
cj := make(chan int, 0)         // unbuffered channel of integers
cs := make(chan *os.File, 100)  // buffered channel of pointers to Files

创建channel时可以提供一个可选的整型参数，用于设置该channel的缓冲区大小。该值缺省为0，用来构建默认的“无缓冲channel”，也称为“同步channel”。

Channel作为goroutine间的一种通信机制，与操作系统的其它通信机制类似，一般有两个目的：同步，或者传递消息。

同步

c := make(chan int)  // Allocate a channel.
// Start the sort in a goroutine; when it completes, signal on the channel.
go func() {list.Sort()c <- 1  // Send a signal; value does not matter.
}()
doSomethingForAWhile()
<-c   // Wait for sort to finigo; discard sent value.

上面的示例中，在子goroutine中进行排序操作，主goroutine可以做一些别的事情，然后等待子goroutine完成排序。

接收方会一直阻塞直到有数据到来。如果channel是无缓冲的，发送方会一直阻塞直到接收方将数据取出。如果channel带有缓冲区，发送方会一直阻塞直到数据被拷贝到缓冲区；如果缓冲区已满，则发送方只能在接收方取走数据后才能从阻塞状态恢复。

消息传递

我们来模拟一下经典的生产者－消费者模型。

func Producer (queue chan<- int){for i:= 0; i < 10; i++ {queue <- i}
}func Consumer( queue <-chan int){for i :=0; i < 10; i++{v := <- queuefmt.Println("receive:", v)}
}func main(){queue := make(chan int, 1)go Producer(queue)go Consumer(queue)time.Sleep(1e9) //让Producer与Consumer完成
}

上面的示例在Producer中生成数据，在Consumer中处理数据。

Server编程模型

在server编程，一种常用的模型：主线程接收请求，然后将请求分发给工作线程，工作线程完成请求处理。用go来实现，如下：

func handle(r *Request) {process(r)  // May take a long time.
}func Serve(queue chan *Request) {for {req := <-queuego handle(req)  // Don't wait for handle to finigo.}
}

一般来说，server的处理能力不是无限的，所以，有必要限制线程（或者goroutine）的数量。在C/C++编程中，我们一般通过信号量来实现，在go中，我们可以通过channel达到同样的效果：

var sem = make(chan int, MaxOutstanding)func handle(r *Request) {sem <- 1    // Wait for active queue to drain.process(r)  // May take a long time.<-sem       // Done; enable next request to run.
}func Serve(queue chan *Request) {for {req := <-queuego handle(req)  // Don't wait for handle to finigo.}
}

我们通过引入sem channel，限制了同时最多只有MaxOutstanding个goroutine运行。但是，上面的做法，只是限制了运行的goroutine的数量，并没有限制goroutine的生成数量。如果请求到来的速度过快，会导致产生大量的goroutine，这会导致系统资源消耗完全。

为此，我们有必要限制goroutine的创建数量：

func Serve(queue chan *Request) {for req := range queue {sem <- 1go func() {process(req) // Buggy; see explanation below.<-sem}()}
}

上面的代码看似简单清晰，但在go中，却有一个问题。Go语言中的循环变量每次迭代中是重用的，更直接的说就是req在所有的子goroutine中是共享的，从变量的作用域角度来说，变量req对于所有的goroutine，是全局的。

这个问题属于语言实现的范畴，在C语言中，你不应该将一个局部变量传递给另外一个线程去处理。有很多解决方法，这里有一个讨论。从个人角度来说，我更倾向下面这种方式：

func Serve(queue chan *Request) {for req := range queue {sem <- 1go func(r *Request) {process(r)<-sem}(req)}
}

至少，这样的代码不会让一个go的初学者不会迷糊，另外，从变量的作用域角度，也更符合常理一些。

在实际的C/C++编程中，我们倾向于工作线程在一开始就创建好，而且线程的数量也是固定的。在go中，我们也可以这样做：

func handle(queue chan *Request) {for r := range queue {process(r)}
}func Serve(clientRequests chan *Request, quit chan bool) {// Start handlersfor i := 0; i < MaxOutstanding; i++ {go handle(clientRequests)}<-quit  // Wait to be told to exit.
}

开始就启动固定数量的handle goroutine，每个goroutine都直接从channel中读取请求。这种写法比较简单，但是不知道有没有“惊群”问题？有待后续分析goroutine的实现。

传递channel的channel

channel作为go语言的一种原生类型，自然可以通过channel进行传递。通过channel传递channel，可以非常简单优美的解决一些实际中的问题。

在上一节中，我们主goroutine通过channel将请求传递给工作goroutine。同样，我们也可以通过channel将处理结果返回给主goroutine。

主goroutine：

type Request struct {args        []intresultChan  chan int
}request := &Request{[]int{3, 4, 5}, make(chan int)}
// Send request
clientRequests <- request
// Wait for response.
fmt.Printf("answer: %d\n", <-request.resultChan)

主goroutine将请求发给request channel，然后等待result channel。子goroutine完成处理后，将结果写到result channel。

func handle(queue chan *Request) {for req := range queue {result := do_something()req.resultChan <- result}
}

多个channel

在实际编程中，经常会遇到在一个goroutine中处理多个channel的情况。我们不可能阻塞在两个channel，这时就该select场了。与C语言中的select可以监控多个fd一样，go语言中select可以等待多个channel。

    c1 := make(chan string)c2 := make(chan string)go func() {time.Sleep(time.Second * 1)c1 <- "one"}()go func() {time.Sleep(time.Second * 2)c2 <- "two"}()for i := 0; i < 2; i++ {select {case msg1 := <-c1:fmt.Println("received", msg1)case msg2 := <-c2:fmt.Println("received", msg2)}}

在C中，我们一般都会传一个超时时间给select函数，go语言中的select没有该参数。

select语句

select语句可以用于多个channel的读或者写。它与switch语句比较类似，只不过select只用于channel。如果有多个channel可以处理，那么select随机选择一个channel处理：

for {  // send random sequence of bits to cselect {case c <- 0:  // note: no statement, no fallthrough, no folding of casescase c <- 1:}
}

如果所有channel都不能处理，如果有default语句，则执行default，如果没有default，则会阻塞，直到有channel可以处理。一个处理nil channel，没有default的select会永远阻塞。这常用于daemon程序。

select {}  // block forever

考虑如下代码：

package main
import "fmt"
func main(){fmt.Println("start")select{}
}

上面的代码会返回下面的错误：

$ go run select1_ex.go
start
fatal error: all goroutines are asleep - deadlock!

需要改成下面这种方式：

package main
import "time"
import "fmt"
func main(){fmt.Println("start")go func(){for {time.Sleep(time.Second * 1)fmt.Println("do some work")}}()select{}
}

参考golang spec: Select statements。

超时

由于select本身并不支持超时，我们需要额外的手段来模拟超时：

timeout := make(chan bool, 1)
go func() {time.Sleep(1 * time.Second)timeout <- true
}()select {case <-ch:// a read from ch has occurred
case <-timeout:// the read from ch has timed out
}

我们可以通过一个单独的timeout channel和goroutine来实现超时机制。