协程与通道

并发、并行和协程

协程定义

  • 并发：一个并发程序能在一个处理器或内核使用多个线程执行
  • 并行：一个并行程序在某个时间点能使用多个处理器资源
  • goroutine 是根据一个或以上的线程的可用性，映射在它们之上；协程调度器在 Go 运行的时候就能完成对协程的调度。
  • goroutine 和 coroutine 的区别：
    • goroutine 意味着并行，协程一般来说并不能这样
    • goroutine 通过信道来相互通信；coroutine 通过让出和恢复操作来通信

协程间的信道

channel 的创建

1
2
3
4
5
6
// var identifier chan datatype
var a chan string
a = make(chan string)

// OR
a := make(chan string)

通信操作符 <-

  • 发送
    ch <- int1 用通道 ch 发送变量 int1

  • 接收
    int2 = <- ch

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
package main

import (
    "fmt"
    "time"
)

func main() {
    ch := make(chan string)

    go setData(ch)
    go getData(ch)

    time.sleep(1e9)
}

func setData(ch chan string) {

    ch <- "test"
    ch <- "hell"

}

func getData(ch chan string) {
    var input string
    for {
        input = <-ch
        fmt.Printf("%s, ", input)
    }
}

由上述代码可以看出，需要通信的两个协程，需要设置同一个 channel 来让两个协程之间进行通信。

通道阻塞

对于同一个 channel，接收操作是阻塞的，直到发送者可用，如果通道中没有数据，接受者阻塞。

通道工厂模式

不将通道作为参数传递给协程，而用函数来生成一个通道并返回（工厂角色）；函数内部有两个匿名函数被协程调用。

使用 select 切换协程

从不同的并发执行的协程中获取值可以通过关键字 select 来实现。

1
2
3
4
5
6
7
8
select {
    case u: <- ch1:
        // ...
    case v: <- ch2:
        // ...
default:
    // ...
}

协程和恢复

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
func server(workChan <-chan *Work) {
    for work:= range workChan {
        go safeDo(work)
    }
}

func safeDo(work *Work){
    defer func(){
        if err := recover(); err!= nil{
            log.Printf("Work failed with %s in %v", err, work)
        }
    }()
    do(work)
}

惰性生成器的实现

生成器指的是当被调用的时候返回一个序列中的下一个结果的函数，例如

1
2
3
generatorInteger() => 1
generatorInteger() => 2
generatorInteger() => 3

生成器每次返回的是序列中的下一个值而非整个序列；这种特性也称之为惰性求值，只在你需要的时候求值，同时保留相关变量资源（内存和 CPU）。例如：生成一个无限数量的偶数序列，如果直接返回一个序列并且一个个的使用，十分浪费内存空间。但是一个含有通道和 Go 协程的函数能实现这一个需求。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
package main

import "fmt"

var resume chan int

func integers() chan int {
	yield := make(chan int)
	count := 0

	go func() {
		for {
			yield <- count
			count++
		}
	}()
	return yield
}

func generateInteger() int {
	return <-resume
}

func main() {
	resume = integers()
	fmt.Println(generateInteger())
	fmt.Println(generateInteger())
	fmt.Println(generateInteger())
}

如果在程序中埋点，会发现其实从通道中读取的部分值是稍早时候产生，而并不是在程序被调用的时候产生的，在一般情况下并不需要考虑这些。

不过如果确实需要用到这个特性，需要实现一个请求-响应机制。

当生成器生成数据的过程是计算密集型且各个结果的顺序并不重要的时候，那么就可以将生成器放入到 go 协程实现并行化。但是，大量的使用 go 协程带来的开销可能要远远大于收益。

以上，通过使用空接口、闭包和高阶函数，我们能实现一个通用的惰性生成器的工厂函数 BuildLazyEvaluator。 工厂函数需要一个函数和一个初始状态来作为输入参数，返回一个无参数、返回值是生成序列的函数。传入的函数需要计算出下一个返回值以及下一个状态函数。在工厂函数中，创建一个通道和无线循环的 go 协程。返回值被放到了通道中，返回函数稍后被调用的时候从通道中获取该返回值。每当取到一个值的时候，下一个值即被计算。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
package main

import "fmt"

type Any interface{}
type EvalFunc func(Any) (Any, Any)

func main() {
	evenFunc := func(state Any) (Any, Any) {
		os := state.(int)
		ns := os + 2
		return os, ns
	}

	even := BuildLazyIntEvaluator(evenFunc, 0)

	for i := 0; i < 20; i++ {
		fmt.Printf("%vth even: %v\n", i, even())
	}
}

func BuildLazyEvaluator(evalFunc EvalFunc, initState Any) func() Any {
	retValChan := make(chan Any)
	loopFunc := func() {
		var actState Any = initState
		var retVal Any
		for {
			retVal, actState = evalFunc(actState)
			retValChan <- retVal
		}
	}
	retFunc := func() Any {
		return <-retValChan
	}

	go loopFunc()
	return retFunc
}

func BuildLazyIntEvaluator(evalFunc EvalFunc, initState Any) func() int {
	ef := BuildLazyEvaluator(evalFunc, initState)
	return func() int {
		return ef().(int)
	}
}