Golang超全面讲解并发

func main() { for i := 0; i <10; i++ { //开启并发打印 go func(i int) { fmt.Printf("hello goroutine : %d \n", i) }(i) } time.Sleep(time.Millisecond) } 

func main() { var a [10]int for i := 0; i <10; i++ { //如果这里不将i传入参数直接引用外部的参数会出现：数据冲突（race condition） go func(i int) { // 打印语句会进行协程的调度，会交出控制权：fmt.Printf("hello goroutine : %d \n", i) a[i]++ //通过 Gosched() 可以手动交出协程的控制权；如果不写这个语句协程就不会交出控制权进行调度执行，就会一直卡死在这里 runtime.Gosched() }(i) } time.Sleep(time.Millisecond) fmt.Println(a) } 

func chanDemo() { //var c chan int //为 nil 的chan不能使用 c := make(chan int) go func() { for  { n := <- c fmt.Println(n) } }() c <- 1 //向 c 里面发送数据（在发送时需要 goroutine 来进行接收，否则就会死锁） c <- 2 } 

func worker(i int, c chan int) { for  { fmt.Printf("Worker :%d, accpet: %c \n", i, <-c) } } func chanDemo() { var channels [10]chan int for i := 0; i <10; i++ { channels[i] = make(chan int) go worker(i, channels[i]) } for i := 0; i <10; i++ { channels[i] <- 'a' + i } for i := 0; i <10; i++ { channels[i] <- 'A' + i } }

// chan<- int :代表只能发数据； <-chan int:代表只能收数据 func worker(i int) chan int { c := make(chan int) go func() { for  { fmt.Printf("Worker :%d, accpet: %c \n", i, <-c) } }() return c } func chanDemo() { var channels [10]chan int for i := 0; i <10; i++ { channels[i] = worker(i) } for i := 0; i <10; i++ { channels[i] <- 'a' + i } for i := 0; i <10; i++ { channels[i] <- 'A' + i } }

func chanClose() { //创建一个缓冲区 c := make(chan int, 3) c <- 1 c <- 2 c <- 3 //方法一： go func() { for  { //如果没有接收到数据就直接返回 if i, ok := <-c; !ok { fmt.Println(" close channel.....") break } else { //死循环读取，如果外部 chan 已经关闭了，这里会一直接收具体数据的默认值 fmt.Printf("%d\n", i) } } }() //方法二： go func() { for n := range c { //如果没有接收到数据就直接返回 fmt.Printf("%d\n", n) } }() //close关闭 channel close(c) } 

func ChanDemo() { workers := make([]worker, 10) for i, _ := range workers { workers[i] = createWorker(i) } for i, worker := range workers{ worker.in <- 'a' + i } for i, worker := range workers{ worker.in <- 'A' + i } //这种方式可以接收消费者传入的数据，如果消费者传入的数据是同步的话，这里也会出现死锁 for _, worker := range workers { <- worker.done <- worker.done } } //定义一个worker的结构用来存放 chan数据 type worker struct { done chan bool in chan int } func createWorker(id int) worker  { w := worker{ done: make(chan bool), in: make(chan int), } go doWorker(id, w.in, w.done) return w } func doWorker(id int, in chan int, done chan bool)  { for i := range in { fmt.Printf("Worker:%d, accept:%c \n", id, i) go func() { //异步发送，否则会出现死锁，因为发送会加锁 done <- true }() } }

//定义一个worker的结构用来存放 chan数据 type done2Worker struct { in chan int //使用指针传递 wg *sync.WaitGroup } func create2Worker(id int, group *sync.WaitGroup) done2Worker  { w := done2Worker{ wg: group, in: make(chan int), } go do2Worker(id, w) return w } func do2Worker(id int, worker2 done2Worker)  { for i := range worker2.in { fmt.Printf("Worker:%d, accept:%c \n", id, i) //执行完成 worker2.wg.Done() } } func Chan2Demo() { var wg sync.WaitGroup workers := make([]done2Worker, 10) for i, _ := range workers { workers[i] = create2Worker(i, &wg) } wg.Add(20) for i, worker := range workers { worker.in <- 'a' + i } for i, worker := range workers { worker.in <- 'A' + i } wg.Wait() }

func generator() chan int { out := make(chan int) go func() { i := 0 for { //休眠随机数 time.Sleep(time.Duration(rand.Intn(1500)) * time.Millisecond) out <- i i++ } }() return out } func main() { //非阻塞式获取数据，谁先出数据就执行哪一段逻辑 var c1, c2 = generator(), generator() // 10秒钟后发送一次数据 after := time.After(time.Second * 10) //每秒钟都会写一次数据 tick := time.Tick(time.Second) //死循环获取channel中的数据 for  { select { //会通过select关键字进行调用，谁先来数据，就执行谁 case n := <-c1: fmt.Println("Received from c1:", n) case n := <-c2: fmt.Println("Received from c2:", n) case <-after: fmt.Println("ten second after......") case <-tick: fmt.Println("tick task exec .....") default: fmt.Println("No value received") } } } 

type atomicInt struct { a int //定义互斥量进行同步 lock sync.Mutex } func (a *atomicInt) add()  { a.lock.Lock() defer a.lock.Unlock() a.a++ } func main() { a := atomicInt{ a: 0, } a.add() go a.add() time.Sleep(time.Millisecond) fmt.Println("value : ", a.a) } 

//传入多个chan，返回一个只能输出的chan func fanIn(chs...chan string) <-chan string { //创建管道 c := make(chan string) //这里循环读取 chs 管道传入的数据 for _ , ch := range chs { go func(in chan string) { for  { //循环从chCopy里面读取数据后传入到返回出去的chan //这里不能直接使用 ch ，因为该变量是一个闭包，后续遍历的管道会将其覆盖 c <- <-in } }(ch) } return c } //创建一个channel，循环的发送数据 func msgGen(serviceName string) chan string { ch := make(chan string) go func() { for  { ch<- fmt.Sprintf("hello：%s", serviceName) } }() return ch } func main() { s1 := msgGen("service1") s2 := msgGen("service2") s3 := msgGen("service3") //可以拿到返回出来的 channel 跟服务继续做交互 m := fanIn(s1, s2, s3) for  { fmt.Println(<-m) } }

// ChannelCreateFunc 创建接口，需要传入管道，以及参数 type ChannelCreateFunc interface { // Create 创建函数传入一个任何类型的管道，后面参数选择性传入 Create(ch <-chan any, V...any) (any, bool) } func Creator(c ChannelCreateFunc, ch <-chan any) { if r, ok := c.Create(ch, time.Duration(time.Second)); ok { fmt.Println(r) } else { fmt.Println("未接收到数据") } } func main() { s1 := msgGen("service1") m := fanIn(s1) Creator(timeout.TimeoutCreator{}, m) Creator(noblock.NotBlockCreator{}, m) } 

type NotBlockCreator struct { } func (n NotBlockCreator) Create(ch <-chan any, V...any) (any, bool) { select { case m := <-ch: return m, true default: return "", false } } 

type TimeoutCreator struct { } func (t TimeoutCreator) Create(ch <-chan any, V...any) (any, bool) { size := len(V) if size == 1 { var timeoutValue = V[0] switch v := timeoutValue.(type) { case time.Duration: for  { select { case m := <-ch: return m, true case <-time.After(v): fmt.Println("数据超时接收，直接返回false") return "", false } } } } return "", false } 

// Fscanf函数在读取文件时，遇到\r为默认替换为0，读取\n结束，如果编码不对，这里读取就会出问题 func readMaze(path string) [][]int { file, err := os.Open(path) if err != nil { panic(err) } var row, col int //这里需要取地址，函数里面会更改row和col的值 fmt.Fscanf(file, "%d %d", &row, &col) fmt.Printf("%d\t%d\n", row, col) //创建一个二位数组，一共有多少行 maze := make([][]int, row) for i := range maze { //创建列 maze[i] = make([]int, col) for j := range maze[i] { fmt.Fscanf(file, "%d", &maze[i][j]) } } return maze } func main() { //读取迷宫文件 maze := readMaze("maze/maze.in") for _, row := range maze { for _, col := range row { fmt.Printf("%d\t", col) } fmt.Println() } } 

//点位的结构体 type point struct { i, j int } //定义需要探索的方向 var dirs = [4]point { //当前位置-1，就是向上 {-1, 0}, //左边的点位 {0, -1}, //向下的点位 {1, 0}, //向右的点位 {0, 1}, } //将两个点位相加，就可以获取到下一个点位 func (p point) add(r point) point { return point{p.i + r.i, p.j + r.j} } func (p point) at(grid [][]int) (int, bool) { //首先判断点位是否越界了，例如传入的点位 (-1,0)或者(1, -1) if p.i <0 || p.i>= len(grid) { return 0, false } //判断j列是否越界了 if p.j <0 || p.j>= len(grid[p.i]) { return 0, false } //返回数据 return grid[p.i][p.j], true } // walk 传入迷宫，指定迷宫开始的点位，以及出口的点位 func walk(maze [][]int, start, end point) [][]int  { //创建走过的步 steps := make([][]int, len(maze)) for i := range steps { steps[i] = make([]int, len(maze[i])) } //创建需要探索的队列，初始的点位(0,0) Q := []point{start} for len(Q) > 0 { cur := Q[0] //截取出队列中的头部 Q = Q[1:] //判断如果点位等于出口的点位，那么直接退出 if cur == end { break } //dirs为点位周边的四个方向，我这里采用的是 上、左、下、右 的方向进行探索 for _, dir := range dirs { //将当前点位跟四个方向相加，例如 (0,0) 向上的方向就是(-1,0)，将i的值进行减1 next := cur.add(dir) //判断向上的点位不能超出迷宫的界限，并且返回在迷宫中的值，因为如果返回的值为1，就证明是墙 val, ok := next.at(maze) if !ok || val == 1 { continue } //不等0就证明是墙 val, ok = next.at(steps) if !ok || val != 0 { continue } //如果是起点，就跳过 if next == start { continue } //获取到当前步数的值 curSteps, _ := cur.at(steps) //将走过的点位追加到切片中 steps[next.i][next.j] = curSteps + 1 //继续将下一个点位添加到需要探索的队列当中 Q = append(Q, next) } } return steps }