goroutine与channels

Posted 2020-10-26 K_artorias

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