Golang并发编程之Goroutine及Channel(八)
进程、线程、并行和并发
一个应用程序是运行在机器上的一个进程；进程是一个运行在自己内存地址空间里的独立执行体。一个进程由一个或多个操作系统线程组成，这些线程其实是共享同一个内存地址空间的一起工作的执行体。几乎所有’正式’的程序都是多线程的，以便让用户或计算机不必等待，或者能够同时服务多个请求（如Web服务器），或增加性能和吞吐量（例如，通过对不同的数据集并行执行代码）。
一个并发程序可以在一个处理器或者内核上使用多个线程来执行任务，但是只有在同一个程序在某一个时间点在多个些处理内核或处理器上同时执行的任务才是真正的并行。
并行是一种通过使用多处理器以提高速度的能力。所以并发程序可以是并行的，也可以不是。
公认的使用多线程的应用最主要的问题是内存中的数据共享，它们会被多线程以无法预知的方式进行操作，导致一些无法重现或者随机的结果（称作竞态）。
在Go中，应用程序并发处理的部分被称作goroutines（go协程），它可以进行更有效的并发运算。在协程和操作系统线程之间并无一对一的关系:协程是根据一个或多个线程的可用性，映射（多路复用，执行于）在它们之上的；协程调度器在Go运行时很好的完成了这个工作。
Go从语言层面就支持了并发。同时并发程序的内存管理有时候是非常复杂的，而Go语言提供了自动垃圾回收机制。
Go语言为并发编程而内置的上层API基于CSP(communicating sequential processes)顺序通信进程模型。这就意味着显式锁都是可以避免的，因为Go语言通过安全的通道发送和接受数据以实现同步，这大大的简化了并发程序的编写。
一般情况下，一个普通的桌面计算机跑十几二十几个线程就有点负载过大了，但是同样这台机器却可以轻松的让成百上千甚至过万的goroutine进行资源竞争。
Goroutine简介
Goroutine是一个轻量级的执行线程。这类似我们熟知的线程，但是更加轻量级，可以理解成java中的thread。但是它比线程更小，十几个goroutine可能体现在底层就是五六个线程，Go语言内部帮你实现了这些goroutine之间的内存共享。执行goroutine只需极少的栈内存(大概4~5k),当然会根据相应的数据伸缩。
假设有一个函数调用f(s)。 下面是以通常的方式调用它，同步运行它。
要在goroutine中调用此函数，请使用go f(s)。 这个新的goroutine将与调用同时执行。
也可以为匿名函数调用启动一个goroutine。
在将goroutine之前，我们先看一个串行的例子:
package main
import "fmt"
func loop() {
for i := 0; i < 10; i++ {
fmt.Printf("%d ", i)
}
}
func main() {
loop()
loop()
}
这是最常用的串行方式，也就是先执行第一次loop()方法，然后执行完成后再去执行第二次loop()方法。
执行结果:
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9
但是如果我们把上面main()函数中的第一个loop()放到goroutine中呢?
func main() {
go loop()
loop()
}
执行结果:
0 1 2 3 4 5 6 7 8 9
到这里就看不明白了，为什么只输出了一次0~9，我调用了两次loop()，虽然一次在主线程中，一次在goroutine中，按道理要输出两边0~9，当然这两边的顺序会不一致。
但是我心中的输出结果大体应该是这个样子的010123456273456789789。为什么这里只有一次？
原来，在goroutine还没来得及跑loop的时候，主函数(主协程)已经退出了。
只要主协程退出了，其他所有的子协程也都会退出结束,这就会导致经常会出现这种情况，子协程里面的数据没有执行就退出了
那如何让goroutine执行完毕后告诉主线程我执行完毕了然后再退出呢？ 使用一个信道来告诉主线程即可。
无缓冲的信道在取消息和存消息的时候都会挂起当前的goroutine，除非另一端已经准备好。如果不用信道来阻塞主线的话，主线程就会过早跑完，loop线程都没有机会执行。
无缓冲的信道永远不会存储数据，只负责数据的流通。体现在:
- 从无缓冲信道取数据，必须要有数据流进来才可以，否则当前线阻塞
- 数据流入无缓冲信道, 如果没有其他
goroutine来拿走这个数据，那么当前线阻塞
这里先只贴上代码，具体到channel的部分再讲:
var complete chan int = make(chan int)
func loop() {
for i := 0; i < 10; i++ {
fmt.Printf("%d ", i)
}
complete <- 0 // 执行完毕了，发个消息
}
func main() {
go loop()
<- complete // 直到线程跑完, 取到消息. main在此阻塞住
}
当我们运行这个程序时，首先看到阻塞调用的输出，然后是两个gouroutines的交替输出。 这种交替反映了Go运行时并发运行的goroutine。
package main
import "fmt"
func f(from string) {
for i := 0; i < 3; i++ {
fmt.Println(from, ":", i)
}
}
func main() {
// Suppose we have a function call `f(s)`. Here's how
// we'd call that in the usual way, running it
// synchronously.
f("direct")
// To invoke this function in a goroutine, use
// `go f(s)`. This new goroutine will execute
// concurrently with the calling one.
go f("goroutine")
// You can also start a goroutine for an anonymous
// function call.
go func(msg string) {
fmt.Println(msg)
}("going")
// Our two function calls are running asynchronously in
// separate goroutines now, so execution falls through
// to here. This `Scanln` code requires we press a key
// before the program exits.
var input string
fmt.Scanln(&input)
fmt.Println("done")
}
执行结果是:
direct : 0
direct : 1
direct : 2
goroutine : 0
goroutine : 1
goroutine : 2
going
1123 // 我们键盘输入的
done
Gosched
runtime.Gosched()用于让出CPU时间片，让出当前goroutine的执行权限，调度器安排其他等待的任务执行，并在下次某个时候从该位置恢复执行。
这就像跑步接力赛，A跑了一会碰到代码runtime.Gosched()就把接力棒交给了B,A歇着了，B继续跑。
Goexit
runtime.Goexit()终止当前所在的协程
Go语言并发模型
Go语言中使用了CSP模型来进行线程通信，准确说，是轻量级线程goroutine之间的通信。CSP模型和Actor模型类似，也是由独立的，并发执行的实体所构成，实体之间也是通过发送消息进行通信的。
Actor模型和CSP模型区别:
Actor之间直接通讯，而CSP是通过Channel通讯，在耦合度上两者是有区别的，后者更加松耦合。
主要的区别在于:CSP模型中消息的发送者和接收者之间通过Channel松耦合，发送者不知道自己消息被哪个接收者消费了，接收者也不知道是哪个发送者发送的消息。
在Actor模型中，由于Actor可以根据自己的状态选择处理哪个传入消息，自主性可控性更好些。
在Go语言中为了不堵塞进程，程序员必须检查不同的传入消息，以便预见确保正确的顺序。CSP好处是Channel不需要缓冲消息，而Actor理论上需要一个无限大小的邮箱作为消息缓冲。
CSP模型的消息发送方只能在接收方准备好接收消息时才能发送消息。相反，Actor模型中的消息传递是异步 的，即消息的发送和接收无需在同一时间进行，发送方可以在接收方准备好接收消息前将消息发送出去
通道(channel)
goroutine运行在相同的地址空间，因此访问共享内存必须要做好同步。goroutine奉行通过通信来共享内存，而不是共享内存来通讯。
引用类型channel是CSP模式的具体实现，用于多个goroutine通讯。其内部实现了同步，确保并发安全。
和map类似，channel也是一个对应make创建的底层数据结构的引用。
Go语言中Channel类型的定义格式如下:
ChannelType = ( "chan" | "chan" "<-" | "<-" "chan" ) ElementType .
它包括三种类型的定义。可选的<-代表channel的方向。如果没有指定方向，那么Channel就是双向的，既可以接收数据，也可以发送数据。
chan T // 可以接收和发送类型为 T 的数据
chan <- float64 // 只可以用来发送 float64 类型的数据
<-chan int // 只可以用来接收 int 类型的数据
默认情况下，channel发送方和接收方会一直阻塞直到对方准备好发送或者接收，这就使得Go语言无需加锁或者其他条件，天然支持了并发。
c := make(chan bool) //创建一个无缓冲的bool型Channel
c <- x //向一个Channel发送一个值
<- c //从一个Channel中接收一个值
x = <- c //从Channel c接收一个值并将其存储到x中
x, ok = <- c //从Channel接收一个值，如果channel关闭了或没有数据，那么ok将被置为false
通道是连接并发goroutine的管道。可以从一个goroutine向通道发送值，并在另一个goroutine中接收到这些值。
使用make(chan val-type)创建一个新通道，通道由输入的值传入。使用通道<-语法将值发送到通道。
这里从一个新的goroutine发送ping到在上面的消息通道。
<-channel语法从通道接收值。在这里，将收到上面发送的ping消息并打印出来。当运行程序时，ping消息通过通道成功地从一个
goroutine传递到另一个goroutine。实现了线程间（准确的说goroutine之间的）通信。此属性允许在程序结束时等待ping消息，而不必使用任何其他同步。
package main
import "fmt"
func main() {
// Create a new channel with `make(chan val-type)`.
// Channels are typed by the values they convey.
messages := make(chan string)
// _Send_ a value into a channel using the `channel <-`
// syntax. Here we send `"ping"` to the `messages`
// channel we made above, from a new goroutine.
go func() { messages <- "ping" }()
// The `<-channel` syntax _receives_ a value from the
// channel. Here we'll receive the `"ping"` message
// we sent above and print it out.
msg := <-messages
fmt.Println(msg) // 输出ping
}
默认情况下，通道是未缓冲的，意味着如果有相应的接收(<- chan)准备好接收发送的值，它们将只接受发送(chan <- )。 缓冲通道接受有限数量的值，而没有用于这些值的相应接收器。
这里使一个字符串的通道缓冲多达2个值。因为这个通道被缓冲，所以可以将这些值发送到通道中，而没有相应的并发接收。
之后可以照常接收这两个值。
package main
import "fmt"
func main() {
// Here we `make` a channel of strings buffering up to
// 2 values.
messages := make(chan string, 2)
// Because this channel is buffered, we can send these
// values into the channel without a corresponding
// concurrent receive.
messages <- "buffered"
messages <- "channel"
// Later we can receive these two values as usual.
fmt.Println(<-messages) // buffered
fmt.Println(<-messages) // channel
}
我们可以使用通道在goroutine上同步执行程序。这里有一个使用阻塞接收等待goroutine完成的示例。
这是将在goroutine中运行的函数。done通道将用来通知另一个goroutine这个函数的工作已经完成，发送值以通知已经完成。
启动一个goroutine工作程序，给它一个通知通道。如果从此程序中删除<-done行，程序将在工作程序(worker)启动之前退出。
阻止，直到在通道上收到工作程序的通知。
package main
import "fmt"
import "time"
// This is the function we'll run in a goroutine. The
// `done` channel will be used to notify another
// goroutine that this function's work is done.
func worker(done chan bool) {
fmt.Print("working...")
time.Sleep(time.Second)
fmt.Println("done")
// Send a value to notify that we're done.
done <- true
}
func main() {
// Start a worker goroutine, giving it the channel to
// notify on.
done := make(chan bool, 1)
go worker(done)
// Block until we receive a notification from the
// worker on the channel.
<-done
}
使用make初始化Channel,还可以设置容量:
make(chan int, 100) //创建一个有缓冲的int型Channel
通过channel实现同步
package main
import (
"fmt"
"time"
)
var ch = make(chan int)
func Printer(str string) {
for _, data := range str {
fmt.Printf("%c", data)
time.Sleep(time.Second)
}
fmt.Println("\n")
}
func person1() {
Printer("Hello ")
// 给通道写数据发送
ch <- 666
}
func person2() {
// 从通道中取数据，如果通道没有数据就会堵塞
<- ch
Printer("World")
}
func main() {
go person1()
go person2()
for {
}
}
如果上面不使用channel进行同步的话，就不能完整的打印出Hello World，可能就是打印出Hwoelrlod这样混乱的排序。
Channel缓冲介绍
容量(capacity)代表Channel容纳的最多的元素的数量，代表Channel的缓冲的大小。如果没有设置容量，或者容量设置为0,
说明Channel没有缓冲。
无缓冲通道是指在接收前没有能力保存任何值的通道。这种类型的通道要求发送goroutine和接收goroutine同时准备好，才能完成发送和接收操作。如果两个goroutine没有同时准备好，通道会导致先执行发送或接收操作的goroutine阻塞等待。这种对通道进行发送和接收的交互行为本身就是同步的。其中任意一个操作都无法离开另一个操作单独存在。
在实际开发中，你可以在多个goroutine从/往一个channel中receive/send数据, 不必考虑额外的同步措施。
Channel可以作为一个先入先出(FIFO)的队列，接收的数据和发送的数据的顺序是一致的。
不带缓冲的Channel兼具通信和同步两种特性，在并发开发中颇受青睐。
有缓冲通道:是一种在被接收前能存储一个或者多个值得通道。
这种类型的通道并不强制要求goroutine之间必须同时完成发送和接收。渠道会堵塞发送和接收的条件也会不同。只有在通道中没有要接收的值时，接收动作才会堵塞。只有在通道没有可用缓冲区容纳被发送的值时，发送动作才会堵塞。
Select
Go语言的选择(select)可等待多个通道操作。将goroutine和channel与select结合是Go语言的一个强大功能。
对于这个示例，将选择两个通道。
每个通道将在一段时间后开始接收值，以模拟阻塞在并发goroutines中执行的RPC操作。我们将使用select同时等待这两个值，
在每个值到达时打印它们。
执行实例程序得到的值是one，然后是two。注意，总执行时间只有1〜2秒，因为1-2秒Sleeps同时执行。
package main
import "time"
import "fmt"
func main() {
// For our example we'll select across two channels.
c1 := make(chan string)
c2 := make(chan string)
// Each channel will receive a value after some amount
// of time, to simulate e.g. blocking RPC operations
// executing in concurrent goroutines.
go func() {
time.Sleep(time.Second * 1)
c1 <- "one"
}()
go func() {
time.Sleep(time.Second * 2)
c2 <- "two"
}()
// We'll use `select` to await both of these values
// simultaneously, printing each one as it arrives.
for i := 0; i < 2; i++ {
select {
case msg1 := <-c1:
fmt.Println("received", msg1)
case msg2 := <-c2:
fmt.Println("received", msg2)
}
}
}
执行结果:
received one // 1s后输出
received two // 2s后输出
超时(timeout)
超时对于连接到外部资源或在不需要绑定执行时间的程序很重要。在Go编程中由于使用了通道和选择(select),实现超时是容易和优雅的。
在这个示例中，假设正在执行一个外部调用，2秒后在通道c1上返回其结果。
这里是select实现超时。res：= <-c1等待结果和<-Time。等待在超时1秒后发送一个值。由于选择继续准备好第一个接收，
如果操作超过允许的1秒，则将按超时情况处理。
如果允许更长的超时，如3s，那么从c2的接收将成功，这里将会打印结果。
运行此程序显示第一个操作超时和第二个操作超时。
使用此选择超时模式需要通过通道传达结果。这是一个好主意，因为其他重要的Go功能是基于渠道和Select。
现在看看下面的两个例子：计时器和ticker。
package main
import "time"
import "fmt"
func main() {
// For our example, suppose we're executing an external
// call that returns its result on a channel `c1`
// after 2s.
c1 := make(chan string, 1)
go func() {
time.Sleep(time.Second * 2)
c1 <- "result 1"
}()
// Here's the `select` implementing a timeout.
// `res := <-c1` awaits the result and `<-Time.After`
// awaits a value to be sent after the timeout of
// 1s. Since `select` proceeds with the first
// receive that's ready, we'll take the timeout case
// if the operation takes more than the allowed 1s.
select {
case res := <-c1:
fmt.Println(res)
case <-time.After(time.Second * 1):
fmt.Println("timeout 1")
}
// If we allow a longer timeout of 3s, then the receive
// from `c2` will succeed and we'll print the result.
c2 := make(chan string, 1)
go func() {
time.Sleep(time.Second * 2)
c2 <- "result 2"
}()
select {
case res := <-c2:
fmt.Println(res)
case <-time.After(time.Second * 3):
fmt.Println("timeout 2")
}
}
执行结果:
timeout 1
result 2
通道的基本发送和接收都阻塞。但是，可以使用select和default子句来实现非阻塞发送，接收，甚至非阻塞多路选择(select)。
这里是一个非阻塞接收。如果消息上有可用的值，则选择将使用该值的<-message大小写。如果不是，它会立即采用默认情况。
非阻塞发送工作类似。
可以使用多个上面的默认子句来实现多路非阻塞选择(select)。这里尝试对消息(message)和信号(signals)的非阻塞接收。
package main
import "fmt"
func main() {
messages := make(chan string)
signals := make(chan bool)
// Here's a non-blocking receive. If a value is
// available on `messages` then `select` will take
// the `<-messages` `case` with that value. If not
// it will immediately take the `default` case.
select {
case msg := <-messages:
fmt.Println("received message", msg)
default:
fmt.Println("no message received")
}
// A non-blocking send works similarly.
msg := "hi"
select {
case messages <- msg:
fmt.Println("sent message", msg)
default:
fmt.Println("no message sent")
}
// We can use multiple `case`s above the `default`
// clause to implement a multi-way non-blocking
// select. Here we attempt non-blocking receives
// on both `messages` and `signals`.
select {
case msg := <-messages:
fmt.Println("received message", msg)
case sig := <-signals:
fmt.Println("received signal", sig)
default:
fmt.Println("no activity")
}
}
执行结果:
no message received
no message sent
no activity
关闭通道表示不会再发送更多值。这对于将完成通信到通道的接收器是很有用的。
在这个例子中，我们将使用一个作业通道来完成从main() goroutine到worker goroutine的工作。当没有更多的工作时，则将关闭工作通道。
这里是工作程序goroutine。它反复从j的工作接收more := <-jobs。在这种特殊的2值形式的接收中，如果作业已关闭并且已经接收到通道中的所有值，
则more的值将为false。当已经完成了所有的工作时，使用这个通知。
这会通过作业通道向工作线程发送3个作业，然后关闭它。
package main
import "fmt"
// In this example we'll use a `jobs` channel to
// communicate work to be done from the `main()` goroutine
// to a worker goroutine. When we have no more jobs for
// the worker we'll `close` the `jobs` channel.
func main() {
jobs := make(chan int, 5)
done := make(chan bool)
// Here's the worker goroutine. It repeatedly receives
// from `jobs` with `j, more := <-jobs`. In this
// special 2-value form of receive, the `more` value
// will be `false` if `jobs` has been `close`d and all
// values in the channel have already been received.
// We use this to notify on `done` when we've worked
// all our jobs.
go func() {
for {
j, more := <-jobs
if more {
fmt.Println("received job", j)
} else {
fmt.Println("received all jobs")
done <- true
return
}
}
}()
// This sends 3 jobs to the worker over the `jobs`
// channel, then closes it.
for j := 1; j <= 3; j++ {
jobs <- j
fmt.Println("sent job", j)
}
close(jobs)
fmt.Println("sent all jobs")
// We await the worker using the
// [synchronization](channel-synchronization) approach
// we saw earlier.
<-done
}
执行结果:
sent job 1
sent job 2
sent job 3
sent all jobs
received job 1
received job 2
received job 3
received all jobs
在前面的例子中，我们已经看到了for和range语句如何为基本数据结构提供迭代。还可以使用此语法对从通道接收的值进行迭代。
此范围在从队列接收到的每个元素上进行迭代。因为关闭了上面的通道，迭代在接收到2个元素后终止。
这个示例还示出可以关闭非空信道，但仍然接收剩余值。
package main
import "fmt"
func main() {
// We'll iterate over 2 values in the `queue` channel.
queue := make(chan string, 2)
queue <- "one"
queue <- "two"
close(queue)
// This `range` iterates over each element as it's
// received from `queue`. Because we `close`d the
// channel above, the iteration terminates after
// receiving the 2 elements.
for elem := range queue {
fmt.Println(elem)
}
}
执行结果:
one
two
定时器代表未来的一个事件。可告诉定时器您想要等待多长时间，它提供了一个通道，在当将通知时执行对应程序。在这个示例中，计时器将等待2秒钟。
<-timer1.C阻塞定时器的通道C，直到它发送一个指示定时器超时的值。
如果只是想等待，可以使用time.Sleep。定时器可能起作用的一个原因是在定时器到期之前取消定时器。这里有一个例子。
第一个定时器将在启动程序后约2s过期，但第二个定时器应该在它到期之前停止。
package main
import "time"
import "fmt"
func main() {
// Timers represent a single event in the future. You
// tell the timer how long you want to wait, and it
// provides a channel that will be notified at that
// time. This timer will wait 2 seconds.
timer1 := time.NewTimer(time.Second * 2)
// The `<-timer1.C` blocks on the timer's channel `C`
// until it sends a value indicating that the timer
// expired.
<-timer1.C
fmt.Println("Timer 1 expired")
// If you just wanted to wait, you could have used
// `time.Sleep`. One reason a timer may be useful is
// that you can cancel the timer before it expires.
// Here's an example of that.
timer2 := time.NewTimer(time.Second)
go func() {
<-timer2.C
fmt.Println("Timer 2 expired")
}()
stop2 := timer2.Stop()
if stop2 {
fmt.Println("Timer 2 stopped")
}
}
执行结果为:
Timer 1 expired
Timer 2 stopped
计时器是当想在未来做一些事情，tickers是用于定期做一些事情。 这里是一个例行程序，周期性执行直到停止。
代码机使用与计时器的机制类似:发送值到通道。 这里我们将使用通道上的一个范围内来迭代值，这此值每500ms到达。
代码可以像计时器一样停止。当代码停止后，它不会在其通道上接收任何更多的值。我们将在1600ms后停止。
当运行这个程序时，ticker应该在我们停止之前打3次。
package main
import "time"
import "fmt"
func main() {
// Tickers use a similar mechanism to timers: a
// channel that is sent values. Here we'll use the
// `range` builtin on the channel to iterate over
// the values as they arrive every 500ms.
ticker := time.NewTicker(time.Millisecond * 500)
go func() {
for t := range ticker.C {
fmt.Println("Tick at", t)
}
}()
// Tickers can be stopped like timers. Once a ticker
// is stopped it won't receive any more values on its
// channel. We'll stop ours after 1600ms.
time.Sleep(time.Millisecond * 1600)
ticker.Stop()
fmt.Println("Ticker stopped")
}
执行结果:
Tick at 2018-04-10 14:18:44.909170271 +0800 CST m=+0.500687046
Tick at 2018-04-10 14:18:45.409704505 +0800 CST m=+1.001220264
Tick at 2018-04-10 14:18:45.909330121 +0800 CST m=+1.500844866
Ticker stopped
- 邮箱 ：charon.chui@gmail.com
- Good Luck!