goroutine与channels

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的特性就介绍完了，欢迎批评指正，互相学习。