goroutine(协程)

大家都知道java中的线程Thread,golang没有提供Thread的功能,但是提供了更轻量级的goroutine(协程),协程比线程更轻,创办一个协程很简单,只需要go关键字加上要运行的函数,就可以实现了。看个简单的例子:

package main

import "fmt"

func f(from string) {
for i := 0; i < 3; i++ {
fmt.Println(from, ":", i)
}
} func main() { f("direct") go f("goroutine") go func(msg string) {
fmt.Println(msg)
}("going") fmt.Scanln()
fmt.Println("done")
}

运行结果如下:

direct : 0
direct : 1
direct : 2
goroutine : 0
going
goroutine : 1
goroutine : 2
<enter>
done

Channels(信道)

channels是并发的goroutines之间通信的管道,我们可以从一个goroutine输出一条数据到一个channel,然后用另外一个goroutine读取,这就实现了goroutines之间的通信。下面的例子演示的是主goroutine与另一个goroutine之间的通信:

package main

import "fmt"

func main() {

    messages := make(chan string)

    go func() { messages <- "ping" }()

    msg := <-messages
fmt.Println(msg)
}

channel buffer

channel的默认空间是0,如果向一个默认长度的channel输入一条数据,程序就会阻塞,直到这条数据被读出。但是我们可以在定义channel时设置其buffer的大小,比如:

messages := make(chan string, 2)

上边这条语句定义的channel的缓存大小为2,当输入小于等于2条数据时,程序不会阻塞,但是当输入第三条数据时就会阻塞,直到被读出一条。

channel Synchronization

所谓channel Synchronization,其实就是利用channel的阻塞机制,当想让程序阻塞时,就向一个buffer为0的channel send一条数据,程序就会阻塞从而锁住,当想解锁时,就再把这个channel中的数据输出就ok了

package main

import "fmt"
import "time" func worker(done chan bool) {
fmt.Print("working...")
time.Sleep(time.Second)
fmt.Println("done") done <- true
} func main() { done := make(chan bool, 1)
go worker(done) <-done
}

channel方向

所谓channel方向,其实指的就是该channel是允许信息输入,还是允许信息输出,普通的channel是既可以输入,也可以输出的,但是我们可以通过设置,让之称为单向的channel,语句如下:

只允许输出 : pings <-chan string
只允许输入 : pongs chan<- string

其实很easy,就是在定义的时候在chan关键字的前方或者后方加上“<-”即可,下面看一个相关的demo

package main

import "fmt"

func ping(pings chan<- string, msg string) {
pings <- msg
} func pong(pings <-chan string, pongs chan<- string) {
msg := <-pings
pongs <- msg
} func main() {
pings := make(chan string, 1)
pongs := make(chan string, 1)
ping(pings, "passed message")
pong(pings, pongs)
fmt.Println(<-pongs)
}

select

在多goroutine多channel情况下,我们不知道哪个goroutine的channel会先结束阻塞状体,我们通过select同时等待多个channel,类似于监听,然后当任何一个channel有消息传出的时候,就会类似于switch case的触发机制一样选择相应的channel读取消息,如果同时有多个消息到达,那么就会随机从一个channel读取消息,如果一直没有消息送到,select会一直阻塞。

package main

import "time"
import "fmt" func main() { c1 := make(chan string)
c2 := make(chan string) go func() {
time.Sleep(1 * time.Second)
c1 <- "one"
}()
go func() {
time.Sleep(2 * time.Second)
c2 <- "two"
}() for i := 0; i < 2; i++ {
select {
case msg1 := <-c1:
fmt.Println("received", msg1)
case msg2 := <-c2:
fmt.Println("received", msg2)
}
}
}

Timeout机制

上边的select功能,如果channel种一直没有消息传出呢,那么select就会一直阻塞,在大多数情况下这不是我们的期望,我们希望有一个阈值,过了这个阈值大小我们可以让select解除阻塞,我们就可以通过timeout机制来对这种情况进行设置。

package main

import "time"
import "fmt" func main() { c1 := make(chan string, 1)
go func() {
time.Sleep(2 * time.Second)
c1 <- "result 1"
}() select {
case res := <-c1:
fmt.Println(res)
case <-time.After(1 * time.Second):
fmt.Println("timeout 1")
} c2 := make(chan string, 1)
go func() {
time.Sleep(2 * time.Second)
c2 <- "result 2"
}()
select {
case res := <-c2:
fmt.Println(res)
case <-time.After(3 * time.Second):
fmt.Println("timeout 2")
}
}

Non-Blocking Channel Operations

上面我们知道了对select设置最长阻塞时间,那么现在我们设置让其不阻塞,只需要在select中添加一个default即可

package main

import "fmt"

func main() {
messages := make(chan string)
signals := make(chan bool) select {
case msg := <-messages:
fmt.Println("received message", msg)
default:
fmt.Println("no message received")
} msg := "hi"
select {
case messages <- msg:
fmt.Println("sent message", msg)
default:
fmt.Println("no message sent")
} select {
case msg := <-messages:
fmt.Println("received message", msg)
case sig := <-signals:
fmt.Println("received signal", sig)
default:
fmt.Println("no activity")
}
}

Closing channel

如果把一个channel close掉,那么就不能向这个channel插入数据,我们可以视为对这个channel的输入任务已经完成,代码如下:

package main

import "fmt"

func main() {
jobs := make(chan int, 5)
done := make(chan bool) go func() {
for {
j, more := <-jobs
if more {
fmt.Println("received job", j)
} else {
fmt.Println("received all jobs")
done <- true
return
}
}
}() for j := 1; j <= 3; j++ {
jobs <- j
fmt.Println("sent job", j)
}
close(jobs)
fmt.Println("sent all jobs") <-done
}

range over channel

在前面的例子里,我们知道,range可以对基本的数据结构进行遍历,channel也可以通过range进行遍历

package main

import "fmt"

func main() {

    queue := make(chan string, 2)
queue <- "one"
queue <- "two"
close(queue) for elem := range queue {
fmt.Println(elem)
}
}

timer and tickers

sleep功能都知道,如果我们想在将来某个时间点执行一个语句,或者每1分钟执行一次某函数,怎么办呢,timer和ticker就能完美解决需求。timer可以设置时间间隔大小,让某个语句到timer所设置的时间间隔后执行,之所以有了sleep还用timer,就是因为timer设置的时间间隔可以提前强制stop掉,如下代码:

package main

import "time"
import "fmt" func main() { timer1 := time.NewTimer(2 * time.Second) <-timer1.C
fmt.Println("Timer 1 expired") 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

因为主协程和go启动的func匿名函数相当于两个协程,所以,主协程中输出正确,但是timer2却没有正确输出,就是因为在主协程中通过timer2.stop()停止了timer2的计时等待,于是主协程先跑完了,并没有跑到另外一个协程。(ps 如果不设置等待,其他协程都不会执行,因为主协程会先跑完然后整个程序就跑完了,其他协程根本来不及跑)

tickers是设置心跳机制,先看一段代码:

ackage 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(500 * time.Millisecond)
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(1600 * time.Millisecond)
ticker.Stop()
fmt.Println("Ticker stopped")
}

类似于一个定时器,可以这样理解:ticker.C是一个特别的channel,这里设置的是每500ms就把当前时间输入到这个channel中去,程序通过go关键字启动了一个新的协程,然后通过for循环读出时间并输出,但是读出来后channel中就没有数据了,会阻塞,知道500ms后程序又自动填充一条时间数据进去,阻塞结束,继续循环,直到主协程结束。

Worker Pools的实现

假设有3个人,同时有5份工作,给三个人分配工作随机,但有一个要求,只有完成了手头工作,才能接下一个,实现代码如下:

package main

import "fmt"
import "time" func worker(id int, jobs <-chan int, results chan<- int) {
for j := range jobs {
fmt.Println("worker", id, "started job", j)
time.Sleep(time.Second)
fmt.Println("worker", id, "finished job", j)
results <- j * 2
}
} func main() { jobs := make(chan int, 100)
results := make(chan int, 100) for w := 1; w <= 3; w++ {
go worker(w, jobs, results)
} for j := 1; j <= 5; j++ {
jobs <- j
}
close(jobs) for a := 1; a <= 5; a++ {
<-results
}
}

由于协程和线程的运行特性一样,是无序的,所以每一次的输出结果都是不同的,下面是某次输出:

worker 1 started  job 1
worker 2 started job 2
worker 3 started job 3
worker 1 finished job 1
worker 1 started job 4
worker 2 finished job 2
worker 2 started job 5
worker 3 finished job 3
worker 1 finished job 4
worker 2 finished job 5

rate limit

go优雅的支持对运行程序的速度限制,通过time.Tick()和channel的阻塞特性,类似于ticker一样对程序限速,如下:

package main

import "time"
import "fmt" func main() { // First we'll look at basic rate limiting. Suppose
// we want to limit our handling of incoming requests.
// We'll serve these requests off a channel of the
// same name.
requests := make(chan int, 5)
for i := 1; i <= 5; i++ {
requests <- i
}
close(requests) // This `limiter` channel will receive a value
// every 200 milliseconds. This is the regulator in
// our rate limiting scheme.
limiter := time.Tick(200 * time.Millisecond) // By blocking on a receive from the `limiter` channel
// before serving each request, we limit ourselves to
// 1 request every 200 milliseconds.
for req := range requests {
<-limiter
fmt.Println("request", req, time.Now())
} // We may want to allow short bursts of requests in
// our rate limiting scheme while preserving the
// overall rate limit. We can accomplish this by
// buffering our limiter channel. This `burstyLimiter`
// channel will allow bursts of up to 3 events.
burstyLimiter := make(chan time.Time, 3) // Fill up the channel to represent allowed bursting.
for i := 0; i < 3; i++ {
burstyLimiter <- time.Now()
} // Every 200 milliseconds we'll try to add a new
// value to `burstyLimiter`, up to its limit of 3.
go func() {
for t := range time.Tick(200 * time.Millisecond) {
burstyLimiter <- t
}
}() // Now simulate 5 more incoming requests. The first
// 3 of these will benefit from the burst capability
// of `burstyLimiter`.
burstyRequests := make(chan int, 5)
for i := 1; i <= 5; i++ {
burstyRequests <- i
}
close(burstyRequests)
for req := range burstyRequests {
<-burstyLimiter
fmt.Println("request", req, time.Now())
}
}

自此,我自学的golang的goroutine和channel的特性就介绍完了,欢迎批评指正,互相学习。

05-14 03:34