go语言--通道

Posted wudiffs

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了go语言--通道相关的知识,希望对你有一定的参考价值。

一、两种并发模型

  • 多线程共享内存,以共享内存的方式来通信。比如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!")
        
    

以上是关于go语言--通道的主要内容,如果未能解决你的问题,请参考以下文章

《Go语言实战》摘录:6.5 并发 - 通道

go语言--通道

Go语言之通道

Go语言并发与通道的运用

Go语言之通道

go语言通道详解