Go语言并发之美

下面生成随机数为例，以让我们做一个会并发执行的随机数生成器。

// 函数rand_generator_1 ，返回 int    funcrand_generator_1() int {             return rand.Int()    }

// 函数rand_generator_2，返回通道(Channel)    funcrand_generator_2() chan int {             // 创建通道             out := make(chan int)             // 创建协程             go func() {                       for {                                //向通道内写入数据，如果无人读取会等待                                out <- rand.Int()                       }             }()             return out    }     funcmain() {             // 生成随机数作为一个服务             rand_service_handler :=rand_generator_2()             // 从服务中读取随机数并打印             fmt.Printf("%d\n",<-rand_service_handler)    }

        上面的这段函数就可以并发执行了rand.Int()。有一点值得注意到函数的返回可以理解为一个“服务”。但我们需要获取随机数据时候，可以随时向这个 服务取用，他已经为我们准备好了相应的数据，无需等待，随要随到。如果我们调用这个服务不是很频繁，一个协程足够满足我们的需求了。但如果我们需要大量访 问，怎么办？我们可以用下面介绍的多路复用技术，启动若干生成器，再将其整合成一个大的服务。

// 函数rand_generator_3 ，返回通道(Channel)    funcrand_generator_3() chan int {             // 创建两个随机数生成器服务             rand_generator_1 := rand_generator_2()             rand_generator_2 := rand_generator_2()                  //创建通道             out := make(chan int)                  //创建协程             go func() {                       for {                                //读取生成器1中的数据，整合                                out <-<-rand_generator_1                       }             }()             go func() {                       for {                                //读取生成器2中的数据，整合                                out <-<-rand_generator_2                       }             }()             return out    }

//一个查询结构体    typequery struct {             //参数Channel             sql chan string             //结果Channel             result chan string    }    //执行Query    funcexecQuery(q query) {             //启动协程             go func() {                       //获取输入                       sql := <-q.sql                       //访问数据库，输出结果通道                       q.result <- "get" + sql             }()    }    funcmain() {             //初始化Query             q :=                       query{make(chan string, 1),make(chan string, 1)}             //执行Query，注意执行的时候无需准备参数             execQuery(q)                  //准备参数             q.sql <- "select * fromtable"             //获取结果             fmt.Println(<-q.result)    }

//建立计数器    sem :=make(chan int, N);    //FOR循环体    for i,xi:= range data {             //建立协程        go func (i int, xi float) {            doSomething(i,xi);                       //计数            sem <- 0;        } (i, xi);    }    // 等待循环结束    for i := 0; i < N; ++i { <-sem }

// Aconcurrent prime sieve    packagemain         // Sendthe sequence 2, 3, 4, ... to channel 'ch'.    funcGenerate(ch chan<- int) {             for i := 2; ; i++ {                      ch<- i // Send 'i' to channel 'ch'.             }    }    // Copythe values from channel 'in' to channel 'out',    //removing those divisible by 'prime'.    funcFilter(in <-chan int, out chan<- int, prime int) {             for {                       i := <-in // Receive valuefrom 'in'.                       if i%prime != 0 {                                out <- i // Send'i' to 'out'.                       }             }    }    // Theprime sieve: Daisy-chain Filter processes.    funcmain() {             ch := make(chan int) // Create a newchannel.             go Generate(ch)      // Launch Generate goroutine.             for i := 0; i < 10; i++ {                       prime := <-ch                       print(prime, "\n")                       ch1 := make(chan int)                       go Filter(ch, ch1, prime)                       ch = ch1             }    }

//共享变量有一个读通道和一个写通道组成    typesharded_var struct {             reader chan int             writer chan int    }    //共享变量维护协程    funcsharded_var_whachdog(v sharded_var) {             go func() {                       //初始值                       var value int = 0                       for {                                //监听读写通道，完成服务                                select {                                case value =<-v.writer:                                case v.reader <-value:                                }                       }             }()    }    funcmain() {             //初始化，并开始维护协程             v := sharded_var{make(chan int),make(chan int)}             sharded_var_whachdog(v)             //读取初始值             fmt.Println(<-v.reader)             //写入一个值             v.writer <- 1             //读取新写入的值             fmt.Println(<-v.reader)    }

        协程和内存一样，是系统的资源。对于内存，有自动垃圾回收。但是对于协程，没有相应的回收机制。会不会若干年后，协程普及了，协程泄漏和内存泄漏一样成为 程序员永远的痛呢？一般而言，协程执行结束后就会销毁。协程也会占用内存，如果发生协程泄漏，影响和内存泄漏一样严重。轻则拖慢程序，重则压垮机器。

        C和C++都是没有自动内存回收的程序设计语言，但只要有良好的编程习惯，就能解决规避问题。对于协程是一样的，只要有好习惯就可以了。

        只有两种情况会导致协程无法结束。一种情况是协程想从一个通道读数据，但无人往这个通道写入数据，或许这个通道已经被遗忘了。还有一种情况是程想往一个通道写数据，可是由于无人监听这个通道，该协程将永远无法向下执行。下面分别讨论如何避免这两种情况。

funcnever_leak(ch chan int) {

         //初始化timeout，缓冲为1

         timeout := make(chan bool, 1)

         //启动timeout协程，由于缓存为1，不可能泄露

         go func() {

                   time.Sleep(1 * time.Second)

                   timeout <- true

         }()

         //监听通道，由于设有超时，不可能泄露

         select {

         case <-ch:

                   // a read from ch hasoccurred

         case <-timeout:

                   // the read from ch has timedout

         }

}

        在并发编程大行其道的今天，对协程和通道的支持成为各个平台比不可少的一部分。虽然各家有各家的叫法，但都能满足协程的基本要求—并发执行和可大量创建。笔者对他们的实现方式总结了一下。