Go channel的使用场景

目录





1 信号通知

经常会有这样的场景,当信息收集完成,通知下游开始计算数据:

import (
	"log"
	"time"
)

func main() {
	isOver := make(chan struct{})
	go func() {
		collectMsg(isOver)
	}()
	<-isOver
	calculateMsg()
}

// 采集
func collectMsg(isOver chan struct{}) {
	log.Println("开始采集")
	time.Sleep(3000 * time.Millisecond)
	log.Println("完成采集")
	isOver <- struct{}{}
}

// 计算
func calculateMsg() {
	log.Println("开始进行数据分析")
}

程序输出:

2021/01/25 09:48:23 开始采集
2021/01/25 09:48:26 完成采集
2021/01/25 09:48:26 开始进行数据分析

如果只是单纯的使用通知操作,那么类型就使用 struct{}。因为空结构体在 go 中是不占用内存空间的,不信你看。

func main() {
	res := struct{}{}
	fmt.Println("占用空间:", unsafe.Sizeof(res))
}

程序输出:

占用空间: 0




2 执行任务超时

在做任务处理的时候,并不能保证任务的处理时间,通常会加上一些超时控制做异常的处理。

func main() {
	select {
	case <-doWork():
		log.Println("任务在规定时间内结束")
	case <-time.After(1 * time.Second):
		log.Println("任务处理超时")
	}
}

func doWork() <-chan struct{} {
	ch := make(chan struct{})
	go func() {
		// 处理耗时任务
		log.Println("开始处理任务")
		time.Sleep(2 * time.Second)
		ch <- struct{}{}
	}()
	return ch
}

程序输出:

2021/01/25 09:58:10 开始处理任务
2021/01/25 09:58:11 任务处理超时




3 生产消费模型

生产者只需要关注生产,而不用去理会消费者的消费行为,更不用关心消费者是否执行完毕。而消费者只关心消费任务,而不需要关注如何生产。

func main() {
	ch := make(chan int, 10)
	go consumer(ch)
	go producer(ch)
	time.Sleep(3 * time.Second)
}

// 一个生产者
func producer(ch chan int) {
	for i := 0; i < 10; i++ {
		ch <- i
	}
	close(ch) // 关闭通道
}

// 消费者
func consumer(ch chan int) {
	for i := 0; i < 5; i++ {
		// 5个消费者
		go func(id int) {
			for {
				item, ok := <-ch
				// 如果等于false 说明通道已关闭
				if !ok {
					return
				}
				fmt.Printf("消费者:%d,消费了:%d\n", id, item)

				time.Sleep(50 * time.Millisecond) // 给别人一点机会不会吃亏
			}
		}(i)
	}
}

程序输出:

消费者:4,消费了:0
消费者:0,消费了:1
消费者:1,消费了:2
消费者:2,消费了:3
消费者:3,消费了:4
消费者:3,消费了:6
消费者:1,消费了:8
消费者:2,消费了:7
消费者:0,消费了:5
消费者:4,消费了:9




4 数据传递

极客上一道有意思的题,假设有4个 goroutine,编号为1,2,3,4。每3秒钟会有一个 goroutine 打印出它自己的编号。现在让你写一个程序,要求输出的编号总是按照1,2,3,4这样的顺序打印。类似下图:
Go channel的使用场景

type token struct{}

func main() {
	num := 4
	var chs []chan token
	// 4 个 channel
	for i := 0; i < num; i++ {
		chs = append(chs, make(chan token))
	}
	// 4 个 协程
	for j := 0; j < num; j++ {
		go worker(j, chs[j], chs[(j+1)%num])
	}
	// 先把令牌交给第一个
	chs[0] <- struct{}{}
	select {}
}

func worker(id int, currentCh chan token, nextCh chan token) {
	for {
		// 对应 work 取得令牌
		token := <-currentCh
		fmt.Println(id + 1)
		time.Sleep(3 * time.Second)
		// 传递给下一个
		nextCh <- token
	}
}

程序输出:

1
2
3
4
1
2
3
4
.
.
.




5 控制并发数

经常会写一些脚本,在凌晨的时候对内或者对外拉取数据,但是如果不对并发请求加以控制,往往会导致 groutine 泛滥,进而打满 CPU 资源。往往不能控制的东西意味着不好的事情将要发生。对于我们来说,可以通过 channel 来控制并发数。

func main() {
	limit := make(chan struct{}, 10)
	jobCount := 100
	for i := 0; i < jobCount; i++ {
		go func(index int) {
			limit <- struct{}{}
			job(index)
			<-limit
		}(i)
	}

	time.Sleep(30 * time.Second)
}

func job(index int) {
	// 耗时任务
	time.Sleep(2 * time.Second)
	fmt.Printf("任务:%d 已完成\n", index)
}

上述代码控制了任意时刻只有10和协程在同时执行

当然了,sync.waitGroup 也可以实现并发协程数控制:

func main() {
	var wg sync.WaitGroup
	jobCount := 100
	limit := 10
	for i := 0; i <= jobCount; i += limit {
		index := 0
		if i-limit >= 0 {
			index = i - limit
		}
		for j := index; j < i; j++ {
			wg.Add(1)
			go func(item int) {
				defer wg.Done()
				job(item)
			}(j)
		}
		wg.Wait()
		fmt.Println("------------")
	}
}

func job(index int) {
	// 耗时任务
	time.Sleep(1 * time.Second)
	fmt.Printf("任务:%d 已完成\n", index)
}




6 自定义互斥锁

可以通过 channel 实现一个小小的互斥锁。通过设置一个缓冲区为1的通道,如果成功地往通道发送数据,说明拿到锁,否则锁被别人拿了,等待他人解锁。

type ticket struct{}

type Mutex struct {
	ch chan ticket
}

// 创建一个缓冲区为1的通道作
func newMutex() *Mutex {
	return &Mutex{ch: make(chan ticket, 1)}
}

// 谁能往缓冲区为1的通道放入数据,谁就获取了锁
func (m *Mutex) Lock() {
	m.ch <- struct{}{}
}

// 解锁就把数据取出
func (m *Mutex) unLock() {
	select {
	case <-m.ch:
	default:
		panic("已经解锁了")
	}
}

func main() {
	mutex := newMutex()
	go func() {
		// 如果是1先拿到锁,那么2就要等1秒才能拿到锁
		mutex.Lock()
		fmt.Println("任务1拿到锁了")
		time.Sleep(1 * time.Second)
		mutex.unLock()
	}()
	go func() {
		mutex.Lock()
		// 如果是2拿先到锁,那么1就要等2秒才能拿到锁
		fmt.Println("任务2拿到锁了")
		time.Sleep(2 * time.Second)
		mutex.unLock()
	}()
	time.Sleep(500 * time.Millisecond)
	// 用了一点小手段这里最后才能拿到锁
	mutex.Lock()
	mutex.unLock()
	close(mutex.ch)
}




注意channel 的哪些操作会引发 panic?


1 关闭一个 nil 值的 channel 会引发 panic

package main

func main() {
  var ch chan struct{}
  close(ch)
}

Go channel的使用场景

2 关闭一个已关闭的 channel 会引发 panic

package main

func main() {
	ch := make(chan struct{})
	close(ch)
	close(ch)
}

Go channel的使用场景

3 向一个已关闭的 channel 发送数据

package main

func main() {
	ch := make(chan struct{})
	close(ch)
	ch <- struct{}{}
}

Go channel的使用场景

上一篇:Golang chan的任务分发和优雅退出


下一篇:Go并发之Channel的简单示例