goroutine和channel

goroutine

多线程

func hello() {
   //fmt.Printf("Hello Goroutine!!\n")
   for i:=0;i<100;i++ {
      fmt.Printf("hello:%d\n",i)
      time.Sleep(time.Millisecond)
   }
}

func main() {
   go hello()    //启动了一个独立的线程,使其与下面的代码交替执行，使之成为一个多线程
   //fmt.Printf("main function\n")
   for i:=0;i<100;i++ {
      fmt.Printf("main:%d\n",i)
      time.Sleep(time.Millisecond)
   }
   time.Sleep(time.Second)   //修复代码，使得主线程退出的时候子线程能执行
}

多个goroutine

func nunmers() {
   for i :=0;i<=5;i++ {
      time.Sleep(time.Millisecond*250)
      fmt.Printf("%d\n",i)
   }
}

func chars() {
   for i:='a';i<='e';i++ {
      time.Sleep(time.Millisecond*400)
      fmt.Printf("%c\n",i)
   }
}

func main() {
   go nunmers()
   go chars()
   time.Sleep(time.Second*3)
}

进程和线程

• 进程：

  • 进程是程序在操作系统中的一次执行过程，系统进行资源分配和调度的一个独立单位

• 线程：

  • 线程是进程的一个执行实体，是CPU调度和分派的基本单位，他是比进程更小的能独立运行的基本单位
• 一个进程可以创建和撤销多个线程，同一个进程中的多个线程之间可以并发执行

并发与并行

• 多线程程序在一个核的CPU上运行，这是并发
• 多线程程序在多个核的CPU上运行，这是并行

协程和线程

• 协程： 独立的栈空间，共享堆空间，调度由用户自己控制，本质上有点类似于用户级线程，这些用户级线程的调度也是自己实现的
• 线程： 一个县城上可以跑多个协程，协程是轻量级的线程

goroutine的调度模型

• M（线程）P（上下文）G（goroutine)

设置golang运行的CPU核心数

func main() {
   cpu := runtime.NumCPU()
   //限制核心数（新版本不用考虑），比如监控程序，可以控制其运行的资源消耗
   //runtime.GOMAXPROCS(1)
   for i:=0;i <=8;i++ {
      go func() {
      }()
   }
   fmt.Printf("%d\n",cpu)
   time.Sleep(time.Second*12)
}

channel

• 不同的goroutine之间如何进行通讯：

  • 全局变量和锁同步： 不推荐
  • Channel： 先进先出的队列
• channel的概念
  • 类似于unix中的管道
  • 先进先出
  • 线程安全，多个goroutine同事访问不需要加锁
  • channel是有类型的，一个证书的channel只能整数

channel的声明

• 引用类型，需要使用make来初始化

var 变量名 chan 类型
var test chan int
var test chan string
var test chan map[string]string
var test chan stu
var test chan *stu

• channel的初始化与读写

func main() {
   //var intChan chan int = make(chan int,1)   有缓冲区的channel
   var intChan chan int = make(chan int)      // 无缓冲区channel
   fmt.Printf("intChanin:%p\n",intChan)
   go func() {
      //写入数据
      intChan <- 100
      fmt.Printf("insert item end\n")
   }()

   go func() {
      //读取数据
      fmt.Printf("start\n")
      time.Sleep(time.Second*3)
      var a int
      a = <- intChan
      fmt.Printf("intChan:%d\n",a)
   }()
   time.Sleep(time.Second*5)
}

goroutine和channel相结合

• 生产者消费者模型

func senData(ch chan string) {
   ch <- "Washinton"
   ch <- "Tripoli"
   ch <- "LongDong"
   ch <- "Beijing"
   ch <- "Tokyo"
}

func getData(ch chan string) {
   var input string
   for {
      input = <- ch
      fmt.Printf("input=%s\n",input)
   }
}

func main() {
   ch := make(chan string,10)
   go  senData(ch)
   go getData(ch)
   time.Sleep(time.Second*1)
}

阻塞channel的情况

- 无写入，空读取会阻塞(空channel)
- 写入的数量超过缓冲区也会阻塞 （channel满了）

func sendChans(ch chan string)  {
   var i int
   for {
      var str string
      str = fmt.Sprintf("stu %d\n",i)
      fmt.Printf("write:%s\n",str)
      //无人消费数据，则阻塞,如果无缓冲区的话，写入也是阻塞的
      ch <- str
      i++
   }
}

func main() {
   ch := make(chan string,10)   //带缓冲区
   //ch := make(chan string)   //无缓冲区
   go sendChans(ch)
   time.Sleep(time.Second*200)
}

channel之间的同步

• 新增退出检测的方式保证同步

func sendChannels(ch chan string,exitCh chan bool) {
   ch <- "AA"
   ch <- "BB"
   ch <- "CC"
   ch <- "DD"
   ch <- "EE"
   close(ch)
   exitCh <- true
}

func getChannels(ch chan string,exitCh chan bool) {
   for {
      // 检查channel关闭
      input,ok := <- ch
      if !ok {
         break
      }
      fmt.Printf("getchannels中的input:%s\n",input)
   }
   exitCh <- true
}

func getChannels2(ch chan string,exitCh chan bool) {
   for {
      // 检查channel关闭
      input,ok := <- ch
      if !ok {
         break
      }
      fmt.Printf("getchannels中的input:%s\n",input)
   }
   exitCh <- true
}

func main() {
   ch := make(chan string)
   exitChan := make(chan bool,3)
   go sendChannels(ch,exitChan)
   go getChannels(ch,exitChan)
   go getChannels2(ch,exitChan)
   //三次检查状态，然后退出,不需要time.sleep，等待其他goroutine退出
   <- exitChan    //取出元素，然后扔掉
   <- exitChan
   <- exitChan
}

• 使用waitGroup的方式确保goroutine的同步

func SendChannels1(ch chan string,waitGroup *sync.WaitGroup) {
   ch <- "AA"
   ch <- "BB"
   ch <- "CC"
   ch <- "DD"
   ch <- "EE"
   close(ch)
   fmt.Printf("send data exited\n")
   waitGroup.Done()    //结束goroutine的标志，然后对标志位的数字-1
}

func GetChannels1(ch chan string,waitGroup *sync.WaitGroup) {
   for {
      input,ok := <- ch
      if !ok{
         break
      }
      fmt.Printf("getchannels中的input:%s\n",input)
   }
   fmt.Printf("get data exited\n")
   waitGroup.Done()
}

func GetChannels2(ch chan string,waitGroup *sync.WaitGroup) {
   for {
      input,ok := <- ch
      if !ok{
         break
      }
      fmt.Printf("getchannels中的input:%s\n",input)
   }
   fmt.Printf("get data2 exited\n")
   waitGroup.Done()
}

func main() {
   var wg sync.WaitGroup
   ch := make(chan string)
   wg.Add(3)  //每执行完一次goroutine就会-1
   go SendChannels1(ch,&wg)
   go GetChannels1(ch,&wg)
   go GetChannels2(ch,&wg)
   wg.Wait()   //等到运行完成之后返回
   fmt.Printf("main goroutine exited\n")
}

• 关闭channel的时候，生产者channel的数据是保存的，不丢失

func main() {
   intChan := make(chan int,10)
   for i:=0;i<10;i++ {
      intChan<- i
   }
   //关闭channel
   close(intChan)
   time.Sleep(time.Second)
   for j:=0;j<10;j++ {
      a := <- intChan
      fmt.Printf("a=%d\n",a)
   }
}

• channel的遍历

func GetChannels2(ch chan string,waitGroup *sync.WaitGroup) {
   //for range会自动判断channel是否关闭
   for v :=range ch {
      fmt.Printf("get data2 %s\n",v)
   }

   fmt.Printf("get data2 exited\n")
   waitGroup.Done()
}

• chan的关闭
  • 使用内置函数close进行关闭，chan关闭之后，for range 遍历chan中已经存在的元素后结束
  • 使用内置函数close进行关闭，chan关闭之后，没有使用for range的写法，需要使用v,ok :=<-chan判断chan是否关闭

chan的只读和只写

需要注意<-是在chan关键字的位置，<-在chan左侧表示只读，在右侧表示只写

func chanPerms() {
   var readOnly <- chan int = make(chan int,100)
   // readOnly <- 100 只读不可写
   var writeOnly chan <- int = make(chan int,10)
   // <- writeOnly   只写不可读
}

• 使用场景： 三方调用只读只写权限控制，防止误操作

对chan进行select操作

func main() {
   var intChan chan int = make(chan int,10)
   var strChan chan string = make(chan string,10)
   var wg sync.WaitGroup
   wg.Add(2)
   // 插入数据
   go func() {
      var count int
      for count < 1000 {
         count++
         select {
         case intChan <- 10:
            fmt.Printf("write to int chan succ\n")
         case strChan <- "hello":
            fmt.Printf("write to str chan succ\n")
         default:
            fmt.Printf("all chan is full\n")
            time.Sleep(time.Second)
         }
      }
      wg.Done()
   }()
   wg.Wait()

   //读取数据
   go func() {
      var count int
      for count < 10000 {
         count ++
         select {
         case a := <- intChan:
            fmt.Printf("read from int chain a:%d\n",a)
         case <- strChan:
            fmt.Printf("read from str chan\n")
         default:
            fmt.Printf("all chan is empty\n")
            time.Sleep(time.Second)
         }
      }
      wg.Done()
   }()
   wg.Wait()
}

有疑问加站长微信联系（非本文作者）