如何使用 Channel

例子来自于Concurrency is not parallelism

Google Search: A fake framework

v1.0

var ( Web = fakeSearch("web") Image = fakeSearch("image") Video = fakeSearch("video") ) type Search func(query string) Result func fakeSearch(kind string) Search { return func(query string) Result { time.Sleep(time.Duration(rand.Intn(100)) * time.Millisecond) return Result(fmt.Sprintf("%s result for %q\n", kind, query)) } } func main() { rand.Seed(time.Now().UnixNano()) start := time.Now() results := Google("golang") elapsed := time.Since(start) fmt.Println(results) fmt.Println(elapsed) } 

关键函数

func Google(query string) (results []Result) { results = append(results, Web(query)) results = append(results, Image(query)) results = append(results, Video(query)) return }

Google 2.0

每个 search, 独立并发.
No locks. No condition variables. No callbacks.

func Google(query string) (results []Result) { c := make(chan Result) go func() { c <- Web(query) } () go func() { c <- Image(query) } () go func() { c <- Video(query) } () for i := 0; i < 3; i++ { result := <-c results = append(results, result) } return }

Google 2.1

如果某个服务比较慢,怎么办?
No locks. No condition variables. No callbacks.

func Google(query string) (results []Result) { c := make(chan Result) go func() { c <- Web(query) } () go func() { c <- Image(query) } () go func() { c <- Video(query) } () timeout := time.After(80 * time.Millisecond) for i := 0; i < 3; i++ { select { case result := <-c: results = append(results, result) case <-timeout: fmt.Println("timed out") return } } return }

Google 3.0 Avoid timeout

No locks. No condition variables. No callbacks.

func First(query string, replicas ...Search) Result { c := make(chan Result) searchReplica := func(i int) { c <- replicas[i](query) } for i := range replicas { go searchReplica(i) } return <-c } func Google(query string) (results []Result) { c := make(chan Result) go func() { c <- First(query, Web1, Web2) } () go func() { c <- First(query, Image1, Image2) } () go func() { c <- First(query, Video1, Video2) } () timeout := time.After(80 * time.Millisecond) for i := 0; i < 3; i++ { select { case result := <-c: results = append(results, result) case <-timeout: fmt.Println("timed out") return } } return }

Google 3.1

上面的例子看起来挺完美,但是存在一个严重的内存泄漏,不知道你看出来没有.
First 中的 searchReplica调用,除了第一个会成功返回以外,其他都不会返回.因为堵塞在 c 上面,从而导致了内存泄漏.
改进也很简单

func First(query string, replicas ...Search) Result { c := make(chan Result,len(replicas)) //看似多分配了资源,但是很快就会收回 searchReplica := func(i int) { c <- replicas[i](query) } for i := range replicas { go searchReplica(i) } return <-c }

经过简单的替换,通过 Go 的并发模型,将一个慢的,顺序执行的,故障敏感的程序改造为了一个快速的,并发的,有冗余的,健壮的程序.

完整的 google 3.1

var ( Web1 = fakeSearch("web") Web2 = fakeSearch("web") Image1 = fakeSearch("image") Image2 = fakeSearch("image") Video1 = fakeSearch("video") Video2 = fakeSearch("video") ) type Search func(query string) Result func fakeSearch(kind string) Search { return func(query string) Result { time.Sleep(time.Duration(rand.Intn(100)) * time.Millisecond) return Result(fmt.Sprintf("%s result for %q\n", kind, query)) } } func main() { rand.Seed(time.Now().UnixNano()) start := time.Now() results := Google("golang") elapsed := time.Since(start) fmt.Println(results) fmt.Println(elapsed) } func First(query string, replicas ...Search) Result { c := make(chan Result,len(replicas)) searchReplica := func(i int) { c <- replicas[i](query) } for i := range replicas { go searchReplica(i) } return <-c } func Google(query string) (results []Result) { c := make(chan Result) go func() { c <- First(query, Web1, Web2) } () go func() { c <- First(query, Image1, Image2) } () go func() { c <- First(query, Video1, Video2) } () timeout := time.After(80 * time.Millisecond) for i := 0; i < 3; i++ { select { case result := <-c: results = append(results, result) case <-timeout: fmt.Println("timed out") return } } return }