Go之Channel

时间 2019-11-19

标签 channel 繁體版

原文原文链接

资料来源

https://golang.org/doc/effective_go.html#concurrency
https://talks.golang.org/2012/concurrency.slide#34
https://speakerdeck.com/kavya719/understanding-channelshtml

Shared by communicating

Do not communicate by sharing memory; instead, share memory by communicating.golang

Goroutine

什么是Goroutine?shell

A goroutine has a simple model: it is a function executing concurrently with other goroutines in the same address space. It is lightweight, costing little more than the allocation of stack space. And the stacks start small, so they are cheap, and grow by allocating (and freeing) heap storage as required.数据结构

Goroutine被分配到OS的多个线程中。因此一个Goroutine 阻塞了，其余的任何继续运行。Goroutine的抽象隐藏了线程建立和管理的复杂性。闭包

在一个函数和方法调用的前面加上go 关键字，就是在一个新的goroutine中调用这个函数或方法。当调用完成的时候，那么goroutine将会安静退出。就像在Unixshell中使用&ide

go list.Sort() // run list.Sort concurrently; don't wait for it.

函数字面值一样能够在goroutine中直接调用函数

func Announce(message string, delay time.Duration) {
    go func() {
        time.Sleep(delay)
        fmt.Println(message)
    }()  // Note the parentheses - must call the function.
}

在Go中，函数字面值是闭包。上面的例子没有价值，由于没有任何方式来通知goroutine已经完成。ui

Channels

和map同样，channels 也是使用make来分配的，make的结果就是对一个底层数据结构的引用。若是额外的参数提供了，那么就是提供给缓冲区的大小。默认大小为0，给一个没有buffer或同步的bufferspa

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主要用来同步，同时也能够用来简单的传输数据。对channel有不少的惯用法，这里给出一个。线程

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

channel的接收这将会永远阻塞 ，直到channel中有数据过来。若是channel是unbuffered，那么发送数据一端将会阻塞，直到接收者可以接收。若是channel有buffer，那么发送数据的一端只会阻塞到数据拷贝到buffer中（这个极短的时间会阻塞）。若是buffer已经满了，那么一直阻塞到接受者可以接收一个数据。

带缓冲的channel

限制流量

一个带缓冲的channel能够用做信号量，用来限制流量，在下面的例子中，到来的请求被传递给handle方法，这个方法传递一个数据到channel，处理这个请求，最后从这个管道中接受这个数据。channel 缓冲区的容量限制了同时调用process的数量。

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 := <-queue
        go handle(req)  // Don't wait for handle to finish.
    }
}

一旦MaxOutstanding的处理process，那么更多的数据向channel插入，就会阻塞。直到有process处理完毕，从channel中释放一个value。这样的设计仍然会有问题，Sever对于每个request都会建立一个goroutine，即便是只能有MaxOutstanding个goroutine来处理。结果是，这个程序在短期，若是有大量的请求进来，那么将有大量资源将会被消费。咱们经过给goroutine建立闸门来限制goroutine的建立。

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

这里的bug，对于for循环中req， req是对在每次迭代中从新利用的，因此req对每一个goroutine都会共享，这不是咱们想要的，咱们想要的是，对于每个goroutine，req都是独一无二的。解决办法能够以下：经过将req的值做为形参来传递。

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 // Create new instance of req for the goroutine.
        sem <- 1
        go func() {
            process(req)
            <-sem
        }()
    }
}

对于goroutine，一般咱们但愿启动固定数量的handle goroutine。

func handle(queue chan *Request) {
    for r := range queue {
        process(r)
    }
}

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

goroutine和channel特色

goroutine 随着main goroutine 的退出而中止运行

package main
import "fmt"
func main()｛
    go func() {
        fmt.Println("hello world")
    }()
}

若是运行上面程序，结果是什么都没有，程序就结束了，缘由是main goroutine退出了。

goroutine 一旦读完本身想要的数据后，退出了，即便写goroutine还在不停的写，程序也没错

package main

import "fmt"
import "time"
import "math/rand"

func main() {
    res := make(chan int)
    go func() {
        joe := boring("Joe")
        ann := boring("Ann")
        for i := 0; i < 5; i++ {
            fmt.Println(<-joe)
            fmt.Println(<-ann)
        }
        fmt.Println("You're both boring; I'm leaving.")
        res <- 1
    }()
     <- res  // 即便在main goroutine中等待很长时间，这个程序也没有错
}
func boring(msg string) <-chan string { // Returns receive-only channel of strings.
    c := make(chan string)
    go func() { // We launch the goroutine from inside the function.
        for i := 0; ; i++ {  // 一直在不停的给channel 写数据。
            c <- fmt.Sprintf("%s %d", msg, i)
            time.Sleep(time.Duration(rand.Intn(1e3)) * time.Millisecond)
        }
    }()
    return c // Return the channel to the caller.
}

channel和其它类型的变量同样是一等公民，能够传到函数中做参数，做为返回值等。

这个例子能够见上面的代码，其返回了一个channel做为数据的生成器

select

当一个goroutine中，你要读或写多个channel，golang怎么处理呢？使用select语句，其像switch语句，可是每个case都是用来通讯的。特色以下：

全部的channels都会被计算（意思是都会查看是否可读可写）
Selection 代码块一直阻塞到有channel可读可写
若是有多个channel可读可写，那么随机选择一个
若是有default语句存在，那么当即执行default语句，即便没有channel可读可写

select {
    case v1 := <-c1:
        fmt.Printf("received %v from c1\n", v1)
    case v2 := <-c2:
        fmt.Printf("received %v from c2\n", v1)
    case c3 <- 23:
        fmt.Printf("sent %v to c3\n", 23)
    default:
        fmt.Printf("no one was ready to communicate\n")
  }

注意这里select的语句只能执行一次，没有一直执行的意思。

常见的pattern

生成器

所谓生成器，就是数据不停的生成，这里利用的channel是第一类值：

package main
import "fmt"
import time
func boring(msg string) <- chan string {
    c := make(chan string)
    go func() { // We launch the goroutine from inside the function.
        for i := 0; ; i++ {
            c <- fmt.Sprintf("%s %d", msg, i)
            time.Sleep(time.Duration(rand.Intn(1e3)) * time.Millisecond)
        }
    }()
    return c // Return the channel to the caller.
}
func main() {
    c := boring("boring!") // Function returning a channel.
    for i := 0; i < 5; i++ {
        fmt.Printf("You say: %q\n", <-c)
    }
    fmt.Println("You're boring; I'm leaving.")
}

利用了channel做为返回值，不停的读其中的数据，注意到goroutine能够在boring函数调用存在，其调用后，仍然存在。

Fan-in

所谓Fan-In，至关于多路选择器，有多个输入channel，选择其中的一个channel的输入，做为输出，具体的模式为

这里的代码利用select对channel的语义：

func fanIn(input1, input2 <-chan string) <-chan string {
    c := make(chan string)
    go func() {
        for {
            select {
            case s := <-input1:  c <- s
            case s := <-input2:  c <- s
            }
        }
    }()
    return c
}

任务队列

func main() {
    ch := make(chain  Task, 3)
    for i := 0 ; i < numWorkers; i++ {
        go workder(ch)
    }
    // Send tasks to workers
    helloTasks := getTasks()
    for _, task := range helloTasks {
        taskCh <- task;
    }
}
func worker(ch) {
    for {
        task := <-taskCh
        process(task)
    }
}