第十章 channel select 总结 -阿里云开发者社区

感觉channel在理解起来还有点费劲的, 尤其是select的使用, 既要可以读数据, 又要可以写数据.

复习一下都学了哪些?然后在重点查一下select的资料

一. channel的定义.

channel的读数据--两种方式

package main
import (
    "fmt"
    "time"
)
func main() {
    a := make(chan int)
    // 定义一个goroutine不停的取数据
    go func() {
        for {
            // 方式一: 使用ok来判断是否有数据, 如果有数据则进行处理
            if aa, ok := <-a; ok {
                fmt.Println(aa)
            }
        }
    }()
    // 加数据
    a <- 1
    a <- 2
    a <- 3
    a <- 4
    a <- 5
    a <- 6
    time.Sleep(time.Second)
}

方式一: 使用ok来判断是否有数据, 如果有数据则进行处理

package main
import (
    "fmt"
    "time"
)
func main() {
    a := make(chan int)
    // 定义一个goroutine不停的取数据
    go func() {
        // 方式二: 从range中循环取数据, 取出来则处理
        for aa := range a{
            fmt.Println(aa)
        }
    }()
    // 加数据
    a <- 1
    a <- 2
    a <- 3
    a <- 4
    a <- 5
    a <- 6
    time.Sleep(time.Second)
}

方式二: 使用range来取数据

channel的写数据

channel的关闭数据

带缓冲的channel

a := make(chan int, 3)

函数是一等公民, channel也是一等公民, channel可以作为参数, 返回值, 数组等

函数的可扩展性

二. select的使用

一个select语句用来选择哪个case中的发送或接收操作可以被立即执行。它类似于switch语句，但是它的case涉及到channel有关的I/O操作。

或者换一种说法，select就是用来监听和channel有关的IO操作，当 IO 操作发生时，触发相应的动作。

select的用法与switch非常类似，由select开始一个新的选择块，每个选择条件由case语句来描述。与switch语句可以选择任何可使用相等比较的条件相比，

select有比较多的限制，其中最大的一条限制就是每个case语句里必须是一个IO操作，确切的说，应该是一个面向channel的IO操作。

“ select”语句的执行分几个步骤进行：

For all the cases in the statement, the channel operands of receive operations and the channel and right-hand-side expressions of send statements are evaluated exactly once, in source order, upon entering the "select" statement.

(所有channel表达式都会被求值、所有被发送的表达式都会被求值。求值顺序：自上而下、从左到右.结果是选择一个发送或接收的channel，无论选择哪一个case进行操作，表达式都会被执行。RecvStmt左侧短变量声明或赋值未被评估.)

If one or more of the communications can proceed, a single one that can proceed is chosen via a uniform pseudo-random selection. Otherwise, if there is a default case, that case is chosen. If there is no default case, the "select"

statement blocks until at least one of the communications can proceed.(如果有一个或多个IO操作可以完成，则Go运行时系统会随机的选择一个执行，否则的话如果有default分支，则执行default分支语句，如果连default都没有，则select语句会一直阻塞，直到至少有一个IO操作可以进行.)

Unless the selected case is the default case, the respective communication operation is executed.(除非选择的情况是default case，否则将执行相应的操作。)

If the selected case is a RecvStmt with a short variable declaration or an assignment, the left-hand side expressions are evaluated and the received value (or values) are assigned.(如果所选case是带有简短变量声明或赋值的RecvStmt，则将评估左侧表达式并分配接收值（或多个值）)

The statement list of the selected case is executed.(所选择的case被执行)

备注:RecvStmt指的就是短变量. 例如: case n := <- a1, 其中的n就是短变量

示例1：select语句会一直等待，直到某个case里的IO操作可以进行

package main
import (
    "fmt"
    "time"
)
func main() {
    a1 := make(chan int)
    a2 := make(chan int)
    // 向管道a1中放数据
    go func() {
        time.Sleep(time.Second * 2)
        a1 <- 1
    }()
    // 向管道a2中放数据
    go func() {
        time.Sleep(time.Second * 5)
        a2 <- 2
    }()
    //使用select从管道a1, a2中读数据
    select {
        case n := <- a1:
            fmt.Println(n)
        case n:= <- a2:
            fmt.Println(n)
    }
    time.Sleep(time.Second * 10)
}

一直等着, 知道2s后a1通道有数据执行第一个case后打印, 5s后a2通道有数据, select执行第二个case取出

这个demo可以更好的说明, select一直在阻塞等待. 知道有符合条件的case执行.

package main
import (
    "fmt"
    "time"
)
func main() {
    a1 := make(chan int)
    a2 := make(chan int)
    // 向管道a1中放数据
    go func() {
        time.Sleep(time.Second * 1)
        a1 <- 1
    }()
    // 向管道a2中放数据
    go func() {
        time.Sleep(time.Second * 6)
        a2 <- 2
    }()
    //使用select从管道a1, a2中读数据
    for {
        select {
            case n := <- a1:
                fmt.Println(n)
                time.Sleep(time.Second)
            case n:= <- a2:
                fmt.Println(n)
                time.Sleep(time.Second * 3)
            default:
                fmt.Println("default")
                time.Sleep(time.Second * 2)
        }
    }
    time.Sleep(time.Second * 10)
}

示例2：select语句的多个case同时满足条件, 执行那个case是随机的

同样是上面的demo, 两个时间都改为2s. 发现, 运行结果, 有时打印的是1, 有时打印的是2

func main() {
    a1 := make(chan int)
    a2 := make(chan int)
    // 向管道a1中放数据
    go func() {
        time.Sleep(time.Second * 2)
        a1 <- 1
    }()
    // 向管道a2中放数据
    go func() {
        time.Sleep(time.Second * 2)
        a2 <- 2
    }()
    //使用select从管道a1, a2中读数据
    select {
        case n := <- a1:
            fmt.Println(n)
        case n:= <- a2:
            fmt.Println(n)
    }
    time.Sleep(time.Second * 4)
}

示例3：所有channel表达式都会被求值、所有被发送的表达式都会被求值。求值顺序：自上而下、从左到右.

package main
import "fmt"
var a1 = make(chan int)
var a2 = make(chan int)
var a = []chan int {a1, a2}
var num = []int {1, 2, 3, 4, 5, 6}
func main() {
    go func(a chan int) {
        fmt.Println(<- a)
    }(a1)
    i:= 1
    select {
    case getChan(i-1, a) <- getNumber(i, num) :
        fmt.Printf("把数字 %d 放在第 %d 通道里", i, i)
    case n := <- getChan(i-1, a):
        fmt.Printf("把数字从通道 %d 里取出来 %d \n", i, n)
    }
}
func getChan(i int, a []chan int) chan int{
    fmt.Println("chan", a[i])
    return a[i]
}
func getNumber(i int, n []int) int {
    fmt.Println("num", n[i])
    return n[i]
}

结果

微信图片_20220510172435.png

在print之前, 执行了两遍getChan, 一遍getNumber

示例4 break关键字结束select

ch1 := make(chan int, 1)
    ch2 := make(chan int, 1)
    ch1 <- 3
    ch2 <- 5
    select {
    case <- ch1:
        fmt.Println("ch1 selected.")
        break
        fmt.Println("ch1 selected after break")
    case <- ch2:
        fmt.Println("ch2 selected.")
        fmt.Println("ch2 selected without break")
    }

很明显，ch1和ch2两个通道都可以读取到值，所以系统会随机选择一个case执行。我们发现选择执行ch1的case时，由于有break关键字只执行了一句

ch1 selected.
Process finished with exit code 0

但是，当系统选择ch2的case时，打印结果为：

ch2 selected.
ch2 selected without break
Process finished with exit code 0

如此就显而易见，break关键字在select中的作用。

示例5: 最后再来看一下我们的那个复杂的demo

package main
import (
    "fmt"
    "math/rand"
    "time"
)
func generator() chan int{
    c := make(chan int)
    i := 0
    go func() {
        for   {
            time.Sleep(time.Duration(rand.Intn(1500)) * time.Millisecond)
            i++
            c <- i
        }
    }()
    return c
}
func createWorker(i int) chan int{
    out := make(chan int)
    go func() {
        for cc := range out {
            time.Sleep(time.Second*3)
            fmt.Printf("管道a%d,  value:%d\n", i, cc)
        }
    }()
    return out
}
func main() {
    var a1 = generator()
    var a2 = generator()
    c := createWorker(0)
    for {
        select {
        case n := <-a1:
            c <- n
        case n := <-a2:
            c <- n
        /*default:
            fmt.Println("default")
            time.Sleep(time.Second)*/
        }
    }
}

这里定义了两个生产者管道, 一个消费者管道, 然后同时工作

接下来想做的事是: 从生成数据的管道中取出数据, 保存到消费管道中,消费数据.

package main
import (
    "fmt"
    "math/rand"
    "time"
)
func generator(num int) chan int{
    c := make(chan int)
    i := 0
    go func() {
        for   {
            time.Sleep(time.Duration(rand.Intn(1500)) * time.Millisecond)
            i++
            fmt.Printf("管道编号: %d, i值:%d \n", num, i)
            c <- i
        }
    }()
    return c
}
func createWorker(i int) chan int{
    out := make(chan int)
    go func() {
        for cc := range out {
            time.Sleep(time.Second*3)
            fmt.Printf("取数据--管道a%d,  value:%d\n", i, cc)
        }
    }()
    return out
}
func main() {
    var a1 = generator(1)
    var a2 = generator(2)
    c := createWorker(0)
    n := 0
    for {
        select {
        case n := <-a1:
        case n := <-a2:
        case c <- n: // 从管道中取出的数据, 放到消费者管道中
        }
    }
}

这样有一个问题, 当消费数据速度慢时,会丢数据. 解决这个问题, 使用一个数组来接收

package main
import (
    "fmt"
    "math/rand"
    "time"
)
func generator(num int) chan int{
    c := make(chan int)
    i := 0
    go func() {
        for   {
            time.Sleep(time.Duration(rand.Intn(1500)) * time.Millisecond)
            i++
            fmt.Printf("管道编号: %d, i值:%d \n", num, i)
            c <- i
        }
    }()
    return c
}
func createWorker(i int) chan int{
    out := make(chan int)
    go func() {
        for cc := range out {
            time.Sleep(time.Second*3)
            fmt.Printf("取数据--管道a%d,  value:%d\n", i, cc)
        }
    }()
    return out
}
func main() {
    var a1 = generator(1)
    var a2 = generator(2)
    c := createWorker(0)
    var values []int
    for {
        var activeValue int
        // 这里重新定义一个channel的原因是: 如果, values为空. 会将一个nil放入到c中. 
        // 然后执行values = values[1:]会报下标越界. 所以定义activeChannel. 初始值时nil. 给一个nil管道放数据, 
        // 不会执行, 会一直等待
        var activechannel chan int
        if len(values) > 0  {
            activeValue = values[0]
            activechannel = c
        }
        select {
        case n := <-a1:
            values = append(values, n)
        case n := <-a2:
            values = append(values, n)
        case activechannel <- activeValue: // 
            values = values[1:]
        /*default:
            fmt.Println("default")
            time.Sleep(time.Second)*/
        }
    }
}

将数组中的数据取出,放入channel的时候, 可以成功. 但是汇报异常.

这里重新定义一个channel的原因是: 如果, values为空. 会将一个nil放入到c中.

然后执行values = values[1:]会报下标越界. 所以定义activeChannel. 初始值时nil. 给一个nil管道放数据,

不会执行, 会一直等待

接下来, 写个定时器, 让程序10秒后自动退出

package main
import (
    "fmt"
    "math/rand"
    "time"
)
func generator(num int) chan int{
    c := make(chan int)
    i := 0
    go func() {
        for   {
            time.Sleep(time.Duration(rand.Intn(1500)) * time.Millisecond)
            i++
            fmt.Printf("管道编号: %d, i值:%d \n", num, i)
            c <- i
        }
    }()
    return c
}
func createWorker(i int) chan int{
    out := make(chan int)
    go func() {
        for cc := range out {
            time.Sleep(time.Second*3)
            fmt.Printf("取数据--管道a%d,  value:%d\n", i, cc)
        }
    }()
    return out
}
func main() {
    var a1 = generator(1)
    var a2 = generator(2)
    c := createWorker(0)
    // 定时发送任务. 返回的是一个时间的channel
    ta := time.After(time.Second * 10)
    var values []int
    for {
        var activeValue int
        // 这里重新定义一个channel的原因是: 如果, values为空. 会将一个nil放入到c中.
        // 然后执行values = values[1:]会报下标越界. 所以定义activeChannel. 初始值时nil. 给一个nil管道放数据,
        // 不会执行, 会一直等待
        var activechannel chan int
        if len(values) > 0  {
            activeValue = values[0]
            activechannel = c
        }
        select {
        case n := <-a1:
            values = append(values, n)
        case n := <-a2:
            values = append(values, n)
        case activechannel <- activeValue: //
            values = values[1:]
        case <- ta: //每10秒从channel中取出数据,
            fmt.Println("bye")
            return
        /*default:
            fmt.Println("default")
            time.Sleep(time.Second)*/
        }
    }
}

我们定义每超过800ms打印一次超时, 这个和上一个定时的区别是: 他是两个case之间输出数据的时间>800. 而上一个的10秒钟是整个程序执行10秒钟

package main
import (
    "fmt"
    "math/rand"
    "time"
)
func generator(num int) chan int{
    c := make(chan int)
    i := 0
    go func() {
        for   {
            time.Sleep(time.Duration(rand.Intn(1500)) * time.Millisecond)
            i++
            fmt.Printf("管道编号: %d, i值:%d \n", num, i)
            c <- i
        }
    }()
    return c
}
func createWorker(i int) chan int{
    out := make(chan int)
    go func() {
        for cc := range out {
            time.Sleep(time.Second*3)
            fmt.Printf("取数据--管道a%d,  value:%d\n", i, cc)
        }
    }()
    return out
}
func main() {
    var a1 = generator(1)
    var a2 = generator(2)
    c := createWorker(0)
    // 定时发送任务. 返回的是一个时间的channel
    ta := time.After(time.Second * 10)
    var values []int
    for {
        var activeValue int
        // 这里重新定义一个channel的原因是: 如果, values为空. 会将一个nil放入到c中.
        // 然后执行values = values[1:]会报下标越界. 所以定义activeChannel. 初始值时nil. 给一个nil管道放数据,
        // 不会执行, 会一直等待
        var activechannel chan int
        if len(values) > 0  {
            activeValue = values[0]
            activechannel = c
        }
        select {
        case n := <-a1:
            values = append(values, n)
        case n := <-a2:
            values = append(values, n)
        case activechannel <- activeValue: //
            values = values[1:]
        case <- ta: //每10秒从channel中取出数据,
            fmt.Println("bye")
            return
        case <- time.After(time.Millisecond * 800):
            fmt.Println("timeout")
        /*default:
            fmt.Println("default")
            time.Sleep(time.Second)*/
        }
    }
}

最后一个, 定时打印每秒钟数组中积压的数据

package main
import (
    "fmt"
    "math/rand"
    "time"
)
func generator(num int) chan int{
    c := make(chan int)
    i := 0
    go func() {
        for   {
            time.Sleep(time.Duration(rand.Intn(1500)) * time.Millisecond)
            i++
            fmt.Printf("管道编号: %d, i值:%d \n", num, i)
            c <- i
        }
    }()
    return c
}
func createWorker(i int) chan int{
    out := make(chan int)
    go func() {
        for cc := range out {
            time.Sleep(time.Second*3)
            fmt.Printf("取数据--管道a%d,  value:%d\n", i, cc)
        }
    }()
    return out
}
func main() {
    var a1 = generator(1)
    var a2 = generator(2)
    c := createWorker(0)
    // 定时发送任务. 返回的是一个时间的channel
    ta := time.After(time.Second * 10)
    // 定时器, 返回的也是一个管道. 每秒往管道中放一个数据
    tick := time.Tick(time.Second)
    var values []int
    for {
        var activeValue int
        // 这里重新定义一个channel的原因是: 如果, values为空. 会将一个nil放入到c中.
        // 然后执行values = values[1:]会报下标越界. 所以定义activeChannel. 初始值时nil. 给一个nil管道放数据,
        // 不会执行, 会一直等待
        var activechannel chan int
        if len(values) > 0  {
            activeValue = values[0]
            activechannel = c
        }
        select {
        case n := <-a1:
            values = append(values, n)
        case n := <-a2:
            values = append(values, n)
        case activechannel <- activeValue: //
            values = values[1:]
        case <- ta: //每10秒从channel中取出数据,
            fmt.Println("bye")
            return
        case <- time.After(time.Millisecond * 800):
            fmt.Println("timeout")
        case <- tick:
            fmt.Println("len:", len(values))
        /*default:
            fmt.Println("default")
            time.Sleep(time.Second)*/
        }
    }
}