package main
import (
"fmt"
"time"
)
//生成数据
func generator() chan int {
out := make(chan int)
go func() {
i := 0
for{
//向out中发送数据
out <- i
i++
}
}()
//开完 goroutine 同时return
return out
}
func createWorker(id int) chan<- int {
c := make(chan int)
go worker(id ,c)
return c
}
//消费数据
func worker (id int ,c chan int){
for n := range c {
//如果 生成数据的速度 > 发送数据出去的速度
time.Sleep(time.Second)
/*
worker:0 received:0
worker:0 received:3530411
worker:0 received:6836647
中间很多数据被冲掉，所以需要需要把接收到的数据存储下来，排队
*/
fmt.Printf("worker:%d received:%d\n",id,n)
}
}
/*通过通信来共享数据*/
func main() {
var c1,c2 =generator() ,generator()
//从c1 c2中接收数据
//n1 := <- c1
//n2 := <- c2
//如果想同时接收，谁来的快就先接收谁 select
worker := createWorker(0)
//n := 0
//存储收到的数据
var values []int
//10秒钟之后会向channel送一个时间
signal := time.After(10 * time.Second)
tick := time.Tick(time.Second) //定时 每隔一段时间送一个值过来
for{
var activeWorker chan<- int //只定义但是没有创建 ,在select 是可以正常运行的，但是不能被select到，因为是阻塞的
//接收到数据后hasValue = true， 将数据发送出去之后就没有数据了hasValue=false
var activeValue int
if len(values) > 0 {
//有值得时候才make channel
activeWorker= worker
activeValue = values[0] //队列中第一个需要被处理的数据
}
select {//从channel中非阻塞的接收或发送数据，使用select
//c1,c2会等待一段时间才会发送数据
case n := <-c1://从c1中进来
//worker = <-n
//invalid operation: <-n (receive from non-chan type int)
//fmt.Println("received from c1",n)
//将接收到的数据存储到value是中
values = append(values,n)
case n := <-c2://从c2中进来
//worker = <- n
//fmt.Println("received from c2",n)
//将接收到的数据存储到value是中
values = append(values,n)
//向worker中送数据
case activeWorker <- activeValue: //处理队列中第一个数据
//将处理的数据从队列中拿掉
values = values[1:]
case <- tick: //每隔一段时间显示队列长度
fmt.Println("queue lenght=",len(values))
/*
queue lenght= 537594
worker:0 received:0
*/
case <- time.After(800*time.Millisecond): //每两次生成数据之间的时间差超过800毫秒
fmt.Println("timeout")
case <-signal: //当10秒之后 收到停止信号
fmt.Println("bye bye")
return
}
}
}
