go语言--通道

Posted 2022-11-27 wudiffs

一、两种并发模型

• 多线程共享内存，以共享内存的方式来通信。比如java在访问共享数据的时候，通过加锁来实现，java.util.concurrent包。

• CSP（communicating sequential processes）并发模型，以通信的方式来共享内存。Go的CSP并发模型是通过goroutine和channel来实现的。

二、通道

原子函数和互斥函数都能工作，但是依靠它们都不会让编写并发程序变得更简单，更不容易出错，或者更有趣。在Go语言里，你可以使用通道来发送和接收需要共享的资源，在goroutine之间同步。

声明通道时，需要指定将要被共享的数据类型，可以以通过通道共享内置类型、命名类型、结构类型和引用类型的值或者指针。

无缓冲的通道：是指接收前没有能力保存任何值的通道，这种类型的通道要求发送goroutine和接收goroutine都准备好，才能完成发送和接受操作。如果没有同时准备好，回导致先执行发送或接收操作的goroutine阻塞等待。

有缓存的通道：是一种指在被接收前能储存一个或多个值的通道，这种类型的通道并不要求goroutine之间必须同时完成发送和接收。只有在通道中没有接收的值时，接收动作才会阻塞。只有在通道没有可用缓存区容纳被发送的值时，发送动作才会阻塞。

无缓存通道工作流程图：

有缓存通道工作流程图：

三、代码示例

无缓存代码示例1：

package main

import "sync"
// wg1用来等待程序结束
var wg1 sync.WaitGroup

// 无缓冲的通道必须配对操作的goroutine出现，否则会一直阻塞
func main() 
    // 计数加1，表示要等待一个goroutine
    wg1.Add(1)
    // 创建一个无缓冲的通道
    c := make(chan string)
    // 启动一个goroutine
    go func() 
        // 延时调用，在函数退出时调用done()来通知main函数工作已完成。
        defer wg1.Done()
        // 接收消息
        value := <-c
        println(value)
    ()
    // 发送消息
    c <- "hello"
    // 等待完成
    wg1.Wait()

无缓存通道代码示例2，实现阻塞生产者消费者模式：

package main

import (
    "fmt"
    "time"
)
// 生产者通过通道发送数据
func produce(p chan<- int) 
    for i := 0; i < 10; i++ 
        p <- i
        fmt.Println("send:", i)
    

// 消费者通过通道接收数据
func consumer(c <-chan int) 
    for i := 0; i < 10; i++ 
        v := <-c
        fmt.Println("receive:", v)
    


func main() 
    // 无缓存数据
    ch := make(chan int)
    // 生产者给ch赋值后，阻塞，知道消费者取出
    go produce(ch)
    // 消费者取出数据后，进行下一次循环，阻塞，知道生产者生产数据
    go consumer(ch)
    // 优先执行主线程，主线程执行完后，立即退出，没有空余时间执行子线程
    time.Sleep(time.Second)

有缓存通道代码示例1：

package main

func main() 

    // 创建带三个缓冲槽的通道
    c := make(chan int, 3)

    // 发送两个消息，缓冲槽区未满，不会阻塞
    c <- 1
    c <- 2

    // 接收两个消息，缓冲区尚有数据，不会阻塞
    println(<-c)
    println(<-c)

有缓存通道代码示例2，模拟多人工作：

package main

import (
    "sync"
    "fmt"
    "math/rand"
    "time"
)

const (
    // 要使用的goroutine的数量
    numberGoroutines = 4
    // 要处理的工作的数量
    taskLoad = 10
)
// wg用来等待程序完成
var wg2 sync.WaitGroup

func main() 
    // 创建一个有缓存的通道来管理工作
    tasks := make(chan string, taskLoad)
    wg2.Add(numberGoroutines)
    // 启动goroutine来处理工作
    for gr := 1; gr <= numberGoroutines ; gr++  
        go worker(tasks, gr)
    

    // 增加一组要完成的工作
    for post := 1; post <= taskLoad; post++ 
        tasks <- fmt.Sprintf("Task : %d", post)
    

    // 当所有工作都处理完时关闭通道，以便所有通道退出
    // 当关闭通道后，goroutine依旧可以从通道接收数据，但是不能再发送数据
    close(tasks)
    // 等待所有工作完成
    wg2.Wait()

    

// worker作为goroutine启动来处理，从有缓存的通道传入数据
func worker(tasks chan string, worker int) 
    // 通知函数已经返回
    defer wg2.Done()
    for 
        // 等待分配工作
        task, ok := <-tasks
        if !ok 
            // 这意味着通道已经空了，并且已被关闭
            fmt.Printf("Worker: %d : shutting down\\n", worker)
            return
        

        // 显示我们开始工作了
        fmt.Printf("Worker: %d : Started %s\\n", worker, task)
        // 随机等一段时间来完成工作
        sleep := rand.Int63n(100)
        time.Sleep(time.Duration(sleep) * time.Millisecond)
        // 显示我们完成工作了
        fmt.Printf("Worker: %d : Completed %s\\n", worker, task)
    

四、通道结合select

1、select详解

select 
case <-chan1:   
// 如果从 chan1 通道成功接收数据，则执行该分支代码case 
chan2 <- 1:   
// 如果成功向 chan2 通道成功发送数据，则执行该分支代码 
default:   
// 如果上面都没有成功，则进入 default 分支处理流程 

• 每个case语句都必须是一个面向通道的操作。
• 多个case并行执行，select选择先操作成功返回的那个case执行，同时返回，随机选择一个。
• 都没返回，进入default分支，不会出现阻塞。
• Chan1通过为空或者chan2通道已满，立即进入default分支。
• 没有default分支，则会阻塞直到某个通道操作成功。

借助?select?语句我们可以在一个协程中同时等待多个通道达到就绪状态。

2、代码示例

select简单代码示例:

package main

import (
    "fmt"
    "math/rand"
)

func main() 
    // 定义三个chan int类型元素的通道数组
    chs := [3]chan intmake(chan int, 1), make(chan int, 1), make(chan int, 1)
    // 随机生成0-2之间的数字
    index := rand.Intn(3)
    // 向通道发送随机数
    chs[index] <- index
    // 哪一个通道中有值，那个对应的分支就会被执行
    select 
    case <-chs[0]:
        fmt.Println("第一个条件分支被选中")
    case <-chs[1]:
        fmt.Println("第二个条件分支被选中")
    case <-chs[2]:
        fmt.Println("第三个条件分支被选中")
    default:
        fmt.Println("没有分支被选中")
    

使用Select+超时实现无阻塞读写:

package main

import (
    "time"
    "fmt"
)
// 使用Select+超时实现无阻塞读写
func main() 

    c1 := make(chan string, 1)
    c2 := make(chan string, 1)

    go func() 
        time.Sleep(time.Millisecond * 100)
        c1 <- "name : james"
    ()

    go func() 
        time.Sleep(time.Millisecond * 100)
        c2 <- "age : 34"
    ()

    for i := 0; i < 3; i++ 
        // 给通道创建容忍时间，定时器功能，default立即返回
        tm := time.NewTimer(time.Second * 5)
        select 
         case msg1 := <- c1:
            fmt.Println(msg1)
         case msg2 := <- c2:
            fmt.Println(msg2)
         case <-tm.C:
            fmt.Println("send data timeout!")