22-Go语言管道(Channel)

多线程同步问题

• 互斥锁
  • 互斥锁的本质是当一个goroutine访问的时候, 其它goroutine都不能访问
  • 这样就能实现资源同步, 但是在避免资源竞争的同时也降低了程序的并发
    性能,程序由原来的并发执行变成了串行
• 打印案例
  • 没有添加互斥锁, 那么两个人都有机会输出自己的内容
  • 添加互斥锁,只有当一个输出完毕的时候,另一个才能输出

package main

import (
    "fmt"
    "sync"
    "time"
)

//创建一个互斥锁
//可以让程序从并发状态变成并行状态
var lock = sync.Mutex{}

//定义一个打印字符的函数
func myprint(str string)  {
    //添加锁
    lock.Lock()
    for _, value := range str {
        time.Sleep(time.Microsecond *300)
        fmt.Printf("%c", value)
    }

    //解锁
    lock.Unlock()
}

//调用者1
func person1()  {
    myprint("hello")
}

//调用者2
func person2()  {
    myprint("world")
}
func main() {
    //开启go程
    go person1()
    go person2()

    //保证主线程不退出,程序不结束
    for  {
        ;
    }
}

生产者与消费者

• 生产者消费者模型
  • 某个模块(函数)负责生产数据, 这些数据由另一个模块来负责处理
  • 一般生产者消费者模型包含三个部分生产者、缓冲区、消费者
  • 没有缓冲区,消费者发生变化, 会直接影响生产者, 耦合性太强
  • 添加缓冲区可以提高效率

---

生产者和消费者资源竞争问题

• 生产者生产产比较慢, 而消费比较快, 就会导致消费者消费到错误数据

package main

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

//定义数组模拟缓冲区
var arr [10]int

//定义模拟生产者函数
func producter()  {
    //定义随机因子
    rand.Seed(time.Now().UnixNano())
    //产生随机数
    for i := 0;i < 10 ;i++  {
        num := rand.Intn(100)
        fmt.Println("生产者生产了", num)
        //将生产的数据放入缓冲区中
        arr[i] = num
        time.Sleep(time.Millisecond * 300)

    }
}

//定义函数模拟消费者
func consumer(){
    for i := 0;i < 10 ;i++  {
        value := arr[i]
        fmt.Println("------消费者消费了",value)
    }
}

func main() {
    // 我们想要的是, 只有生产者生产了, 我们才能消费
    // 注意点: 在多go程中, 如果生产者生产的太慢, 那么消费者就会消费到错误的数据
    go producter()
    // 注意点: 看上去通过给生产者以及消费者同时加锁就能解决, 只有生产完了才能消费
    //         但是取决于谁想执行加锁操作, 所以不完美
    go consumer()

    for  {
        ;
    }
}

• 利用互斥锁解决问题

package main

import (
    "fmt"
    "math/rand"
    "sync"
    "time"
)
//创建一把互斥锁
var lock = sync.Mutex{}

//定义数组模拟缓冲区
var arr [10]int

//定义模拟生产者函数
func producter()  {
    /*
    为什么在生产者和消费者中都上锁之后, 就可以实现生产完再消费?
    因为生产者和消费者中的锁都是同一把, 都是全局变量lock

    调用Lock()函数的作用: 是修改Mutex结构体中的state属性的值, 将它改为一个非0的值
    每次上锁的时候都会判断有没有被锁定, 如果已经锁定就不锁了, 并且不会执行后面的代码
    调用Unlock函数的作用: 是修改Mutex结构体中的state属性的值, 将它改为0
     */

    //上锁
    // 这里锁定的是当前go程, 也就是当前函数
    // 意味着其它的go程不能执行当前的函数, 只有当前锁定这个函数的go程才能执行这个函数
    lock.Lock()
    //定义随机因子
    rand.Seed(time.Now().UnixNano())
    //产生随机数
    for i := 0;i < 10 ;i++  {
        num := rand.Intn(100)
        fmt.Println("生产者生产了", num)
        //将生产的数据放入缓冲区中
        arr[i] = num
        //time.Sleep(time.Millisecond * 300)
    }
    //解锁
    lock.Unlock()
}

//定义函数模拟消费者
func consumer(){
    //上锁
    lock.Lock()
    for i := 0;i < 10 ;i++  {
        value := arr[i]
        fmt.Println("------消费者消费了",value)
    }

    //解锁
    lock.Unlock()
}

func main() {
    // 我们想要的是, 只有生产者生产了, 我们才能消费
    // 注意点: 在多go程中, 如果生产者生产的太慢, 那么消费者就会消费到错误的数据
    go producter()
    // 注意点: 看上去通过给生产者以及消费者同时加锁就能解决, 只有生产完了才能消费
    //         但是取决于谁想执行加锁操作, 所以不完美
    go consumer()

    for  {
        ;
    }
}

---

Go语言管道

管道的基本使用

• Channel的本质是一个队列
• Channel是线程安全的, 也就是自带锁定功能
• Channel声明和初始化
  • var 变量名称 chan 数据类型
  • make(chan 数据类型, 容量)
  • 管道和切片/字典一样,必须创建后才能使用,否则会报错
  • Channel和切片还有字典一样, 是引用类型,是地址传递

package main

import "fmt"

func main() {
    /*
    1.什么是管道:
    管道就是一个队列, 具备先进先出的原则
    是线程安全的, 也就是自带锁定功能

    2.管道作用:
    在Go语言的协程中, 一般都使用管道来保证多个协程的同步, 或者多个协程之间的通讯

    3.如何声明一个管道, 和如何创建一个管道
    管道在Go语言中和切片/字典一样也是一种数据类型
    管道和切片/字典非常相似, 都可以用来存储数据, 都需要make之后才能使用
    3.1管道声明格式:
    var 变量名称 chan 数据类型
    var myCh chan int
    如上代码的含义: 声明一个名称叫做myCh的管道变量, 管道中可以存储int类型的数据
    3.2管道的创建:
    make(chan 数据类型, 容量)
    myCh = make(chan int, 3);
    路上代码的含义: 创建一个容量为3, 并且可以保存int类型数据的管道

    4.管道的使用
    4.1如何往管道中存储(写入)数据?
    myCh<-被写入的数据
    4.2如何从管道中获取(读取)数据?
    <-myCh
    对管道的操作是IO操作
    例如: 过去的往文件中写入或者读取数据, 也是IO操作
    例如: 过去的往屏幕上输出内容, 或者从屏幕获取内容, 也是IO操作
    stdin / stdout / stderr

    注意点:
    和切片不同, 在切片中make函数的第二个参数表示的切片的长度(已经存储了多少个数据),
    而第三个参数才是指定切片的容量
    但是在管道中, make函数的第二个参数就是指定管道的容量, 默认长度就是0
     */

     //1.定义一个管道
     //var myChan chan int
     //2.使用make创建管道
     myChan := make(chan int, 3)
     //3.往管道中存储数据
     myChan<-1
     myChan<-2
     myChan<-3
     //从管道中取出数据
     fmt.Println(<-myChan)
     fmt.Println(<-myChan)
     fmt.Println(<-myChan)
     

    //定义一个管道
    var myChan chan int
    //直接使用管道
    //注意点: 会报错,管道定义完成后不创建是无法直接使用的
    //myChan<-666
    //fmt.Println(<-myChan)

    //创建管道
    myChan = make(chan int, 3)
    //只要往管道中写入了数据, 那么len就会增加
    myChan <- 2
    //fmt.Println("len = ", len(myChan), "cap = ", cap(myChan))
    myChan <- 4
    //fmt.Println("len = ", len(myChan), "cap = ", cap(myChan))
    myChan <- 6
    //fmt.Println("len = ", len(myChan), "cap = ", cap(myChan))
    //注意点: 如果len等于cap, 那么就不能往管道中再写入数据了, 否则会报错
    //myChan <- 8

    //管道未写入数据,使用管道去取数据会报错
    //从管道中取数据,len会减少
    //<-myChan
    fmt.Println(<-myChan)
    fmt.Println("len=",len(myChan),"cap = ", cap(myChan))
    fmt.Println(<-myChan)
    fmt.Println("len=",len(myChan),"cap = ", cap(myChan))
    fmt.Println(<-myChan)
    fmt.Println("len=",len(myChan),"cap = ", cap(myChan))
    //注意点: 取数据个数也不可以超出写入的数据个数,否则会报错
    //fmt.Println(<-myChan)
}

---

管道的遍历和关闭

• 管道遍历推荐两种方式
  • for..range方法遍历
  • for死循环遍历

package main

import "fmt"

func main() {
    /*
    管道的遍历:
    可以使用for循环, 也可以使用 for range循环, 以及死循环来遍历
    但是更推荐使用后两者

    因为在企业开发中, 有可能我们不知道管道中具体有多少条数据, 所以如果利用for循环来遍历, 那么无法确定遍历的次数, 并且如果遍历的次数太多, 还会报错
     */

    //创建一个管道
    myChan := make(chan int, 3)
    //往管道中写入数据
    myChan <- 2
    myChan <- 4
    myChan <- 6
    //注意点: 如果不关闭管道,遍历管道会报错
    close(myChan)
    //第一种方法遍历管道
    //for value := range myChan {
    //  fmt.Println(value)
    //}

    //第二种方法遍历管道
    //注意点: 如果被遍历的管道没有关闭, 那么会报错
    //        如果管道没有被关闭, 那么会将true返回给ok, 否则会将false返回给Ok
    for {
        if v,ok := <-myChan; ok {
            fmt.Println(v)
            fmt.Println(ok)
        }else {
            break
        }
    }


    //注意点: 管道关闭后无法往里面写入数据,会报错,但是可以读取数据不会报错
    myChan := make(chan int,3)
    close(myChan)
    //往管道中写入数据
    //myChan<-1
    //myChan<-2
    //myChan<-3
    //从管道中读取数据
    <-myChan
}

---

管道阻塞现象

• 阻塞现象只会发生在go程中运行才会发生,在普通线程中不会发生这种现象

package main

import "fmt"

//定义一个管道
var myChan = make(chan int,2)

func test()  {
    /*myChan<-1
    myChan<-2
    fmt.Println("管道满之前的代码")
    //这里不会报错, 会阻塞, 等到将管道中的数据读出去之后, 有的新的空间再往管道中写
    myChan<-3
    fmt.Println("管道满之后的代码")*/
    //这里不会报错,会阻塞,等到有人往管道中写入数据之后,有新的数据之后才会读取
    fmt.Println("读取之前的代码")
    <-myChan
    fmt.Println("读取之后的代码")
}

func main() {
    /*
    单独在主线程中操作管道, 写满了会报错, 没有数据去读取也会报错
    只要在go程中操作管道, 无论有没有写满, 无论有没有数据都会发生管道阻塞的现象
    */
    go test()
    for  {
        ;
    }
}

• 利用管道阻塞实现并发变串行

package main

import (
    "fmt"
    "time"
)
//创建一个管道
var myChan = make(chan bool)


func printer(str string)  {
    for _, value := range str {
        fmt.Printf("%c", value)
        time.Sleep(time.Microsecond * 300)
    }
}

func person1()  {
    printer("hello")
    //往管道中写入数据
    //只有printer函数执行完,才会往管道中写入数据
    myChan<-true
}

func person2()  {
    //从管道中读取数据
    //只有管道中有数据才会读取,否则会阻塞
    <-myChan
    printer("world")
}

func main() {
    go person1()
    go person2()
    for  {
        ;
    }
}

• 利用管道阻塞实现生产者和消费者模型

package main

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

//定义管道模拟缓冲区
var myChan = make(chan int, 10)

//定义生产者函数
func producter() {
    //定义随机因子
    rand.Seed(time.Now().UnixNano())
    //生成随机数
    for i := 0; i < 10; i++ {
        num := rand.Intn(100)
        fmt.Println("生产者生产了", num)
        //将生产的数据存入到管道中
        myChan <- num
    }
}

//定义函数模拟消费者
func customer() {
    //从管道中读取数据
    for i := 0; i < 10; i++ {
        num := <-myChan
        fmt.Println("----消费者消费了", num)
    }

}
func main() {
    //创建两个go程
    //多个生产者和多个消费者
    go producter()
    go producter()


    go customer()
    go customer()

    for {
        ;
    }
}

---

无缓冲区管道

• 管道容量为0的管道就是无缓冲区管道

package main

import "fmt"

func main() {
    /*
    // 管道总结:
    // 管道一般都在go程中使用, 不会直接在主线程中使用, 无论是有缓冲的还是没有缓冲的
    // 只要是在go程中使用, 无论是有缓冲的还是没有缓冲的, 都会出现阻塞现象

    */
    //创建无缓冲管道
    // 注意点:
    // 没有缓冲的管道不能直接存储数据
    // 没有缓冲的管道不能直接获取数据
    // 注意点:
    // 想使用没有缓冲的管道, 必须保证读和写同时存在, 而且还必须保证读和写是在不同的go程中,至少有一个在go程中
    // 并且读必须写在写的前面
    //myChan := make(chan int,0)
    // 没有缓冲的管道不能直接获取数据
    //fmt.Println(<-myChan)
    // 没有缓冲的管道不能直接存储数据
    //myChan<-1

    /*//读取管道在go程中
    go func() {
        fmt.Println("读之前的代码")
        fmt.Println(<-myChan)
        fmt.Println("读之前的代码")

    }()

    //写入管道在主线程中
    func(){
        time.Sleep(time.Second * 5)
        fmt.Println("写之前的代码")
        myChan<-2
        fmt.Println("写之后的代码")
    }()*/


    //无缓冲管道单个使用
    //如果是在go程中使用无缓冲的管道, 那么就可以单个使用(只有读, 或者只有写)
    myChan := make(chan int,0)
    go func() {
        //只读不会报错
        //<-myChan
        //只写不会报错
        fmt.Println("写之前的代码")
        myChan<-1
        fmt.Println("写之后的代码")
    }()

    for  {
        ;
    }
}

• 利用无缓冲区管道解决主线程结束问题

package main

import "fmt"

func main() {
    //创建一个有缓冲管道
    myChan := make(chan int,3)
    //创建一个无缓冲管道
    exitChan := make(chan bool,0)

    //往管道中存储数据
    go func() {
        for i := 0; i < 3; i++ {
            myChan<-i
            fmt.Println("生产了",i)
        }
        //无缓冲管道存储数据
        exitChan<-true
    }()

    <-exitChan
    //for  {
    //  ;
    //}
}

---

单向管道

• 默认情况下所有的管道都是双向的管道(可读可写),那么在企业开发中, 我们可能会需要将管道作为函数的参数, 并且还需要限制函数中如何使用管道,,那么这个时候我们就可能会使用单向管道
• 双向管道格式
  • var myCh chan int
• 单向管道格式
  • var myCh chan<- int; 只写的管道
  • var myCh <-chan int; 只读的管道
• 注意点:
  • 双向管道可以转换为单向的管道
  • 但是单向的管道不能转换为双向的管道

package main

import "fmt"

func main() {
    /*
    默认情况下所有的管道都是双向的管道(可读可写)
    那么在企业开发中, 我们可能会需要将管道作为函数的参数, 并且还需要限制函数中如何使用管道,
    那么这个时候我们就可能会使用单向管道

    双向格式:
    var myCh chan int;
    myCh = make(chan int, 5)
    myCh = make(chan int)

    单向格式:
    var myCh chan<- int; 只写的管道
    var myCh <-chan int; 只读的管道

    双向管道可以转换为单向的管道
    但是单向的管道不能转换为双向的管道
    */

    //定义一个双向管道
    myChan := make(chan int,3)
    //定义一个只读的单向管道
    var myChan1 <-chan int
    //定义一个只写的单向管道
    //var myChan2 chan<- int

    /*//将双向管道赋给只读的单向管道
    myChan1 = myChan
    fmt.Println(myChan1)

    //将双向管道赋给只写的单向管道
    myChan2 = myChan
    fmt.Println(myChan2)*/

    //将单向管道赋给双向管道
    //会报错
    //myChan = myChan1
    //fmt.Println(myChan)
}

• 单向管道作为函数参数

package main

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

//定义一个生产者
func producter(buff chan <- int)  {
    //定义随机因子
    rand.Seed(time.Now().UnixNano())
    //生成随机数
    for i := 0;i < 5 ;i++  {
        num := rand.Intn(100)
        //将产生的随机数存储到管道中
        buff <- num
        fmt.Println("生产者生产了", num)
    }
}

//定义消费者函数
func consumer(buff <- chan  int, exitChan chan <- int )  {
    for i := 0;i < 5 ;i++  {
        //读取管道中的数据
        num := <-buff
        fmt.Println("------消费者消费", num)
    }
    //利用管道阻塞解决死循环问题
    exitChan<- 666
}
func main() {
    //定义两个双向管道
    myChan := make(chan int, 5)
    exitChan := make(chan int)

    //开启两个go程
    go producter(myChan)
    go consumer(myChan,exitChan)

    <-exitChan
}

---

select选择结构

• 在企业开发中, 一般情况下使用select都是用于同时消费多个管道中数据
• 在企业开发中, 一般情况下select中的default不用写
• 在企业开发中, 一般情况下使用select来控制退出主线程
• 在企业开发中, 一般情况下使用select来处理超时

package main

import (
    "fmt"
    "time"
)

func main() {
    // 1.创建一个管道
    myCh1 := make(chan int, 5)
    //myCh2 := make(chan int, 5)
    //exitCh := make(chan bool)

    // 2.开启一个协程生产数据
    go func() {
        time.Sleep(time.Second * 5)
        for i := 0; i < 10 ; i++ {
            myCh1<-i
            fmt.Println("生产者1生产了", i)
        }
        close(myCh1)
        //exitCh<-true
    }()
    /*
   go func() {
       time.Sleep(time.Second * 5)
       for i := 0; i < 10 ; i++ {
           myCh2<-i
           fmt.Println("生产者2生产了", i)
       }
       close(myCh2)
   }()

    // 2.在主线程中消费数据
   //for i := 0; i < 10 ; i++ {
   //   num := <-myCh
   //   fmt.Println("------消费者消费了", num)
   //}

   //for num := range myCh{
   //   fmt.Println("------消费者消费了", num)
   //}
    */
    // 注意点: 在企业开发中, 一般情况下使用select都是用于同时消费多个管道中数据
    //         在企业开发中, 一般情况下select中的default不用写
    //         在企业开发中, 一般情况下使用select来控制退出主线程
    //         在企业开发中, 一般情况下使用select来处理超时
    for{
        //fmt.Println("start")
        select {
        case num1 := <-myCh1:
            fmt.Println("------消费者消费了myCh1", num1)
            //case num2 := <-myCh2:
            //  fmt.Println("------消费者消费了myCh2", num2)
            //case <-exitCh:
            //  return
        case <-time.After(3):
            fmt.Println("超时了")
            return
            //default:
            //  fmt.Println("生产者还没有生产好数据")
        }
        //fmt.Println("=====================")
        time.Sleep(time.Millisecond)
    }
    fmt.Println("程序结束了")
}

---

管道是地址传递

package main

import "fmt"

func main() {
    //定义一个双向有缓冲区管道
    var myChan = make(chan int,3)
    fmt.Println(myChan) //0xc00007c080
    fmt.Printf("%p\n", myChan) //0xc00007c080
    fmt.Printf("%p\n", &myChan)  //0xc000072018

    //管道是地址传递
    //定义一个单向管道
    var myChan2 chan <- int
    myChan2 = myChan
    //打印单向管道len和cap
    fmt.Println("len = ", len(myChan2),"cap = ", cap(myChan2))
    //打印双向管道len和cap
    fmt.Println("len = ", len(myChan),"cap = ", cap(myChan))
}

---

定时器

• 对时间的操作方法,一般都在time包中查找

package main

import (
    "fmt"
    "time"
)

func main() {
    /*
    type Timer struct {
        C <-chan Time
        r runtimeTimer
    }
    */

    //1.使用定时器,就要用到time包
    // NewTimer作用, 就是让系统在指定时间之后, 往Timer结构体的C属性中写入当前的时间
    // 让程序阻塞3秒, 3秒之后再执行
    //func NewTimer(d Duration) *Timer

    /*start := time.Now()
    fmt.Println(start)
    //使用定时器
    timer := time.NewTimer(time.Second * 3)
    fmt.Println(<-timer.C)*/

    //2.func After(d Duration) <-chan Time
    //这个定时器底层就是NewTimer实现的
    /*start := time.Now()
    fmt.Println(start)
    //使用定时器
    timer := time.After(time.Second * 3)
    fmt.Println(<-timer)*/

    // 以上的定时器都是一次性的定时器, 也就是只会执行一次
    /*go func() {
        start := time.Now()
        fmt.Println(start)
        timer := time.After(time.Second * 3)
        for  {
            fmt.Println(<-timer)
        }
    }()

    for  {
        ;
    }*/

    //周期性定时器
    start := time.Now()
    fmt.Println(start)
    //定义周期性定时器
    timer := time.NewTicker(time.Second * 2)
    for {
        fmt.Println(<-timer.C)
        timer.Stop()
    }
}