Channel的典型应用场景

2023-11-27 17:54:34

消息交流（生产者和消费者）

一个goroutine可以安全地往Channel中塞数据，另外一个goroutine可以安全地从Channel中读取数据，goroutine就可以安全地实现信息交流了，即生产者和消费者模式。
工作池是一组等待任务并处理任务的协程。

 1 package main
 2 
 3 import (
 4    "fmt"
 5    "math/rand"
 6    "sync"
 7    "time"
 8 )
 9 
10 type Task struct {
11    id        int
12    randomNum int
13 }
14 
15 type Result struct {
16    task               Task
17    randomNumDigitsSum int
18 }
19 
20 var taskCh = make(chan Task, 10)
21 var resultCh = make(chan Result, 10)
22 
23 func createTasks(taskCount int) {
24    for i := 0; i < taskCount; i++ {
25       taskCh <- Task{i + 1, rand.Intn(999)}
26    }
27    close(taskCh)
28 }
29 
30 func getRandomNumDigitsSum(num int) int {
31    sum := 0
32    for num > 0 {
33       sum += num % 10
34       num /= 10
35    }
36    time.Sleep(2 * time.Second)
37    return sum
38 }
39 
40 func work(wg *sync.WaitGroup) {
41    for task := range taskCh {
42       result := Result{task, getRandomNumDigitsSum(task.randomNum)}
43       resultCh <- result
44    }
45    wg.Done()
46 }
47 
48 func createWorkers(workerCount int) {
49    var wg sync.WaitGroup
50    for i := 0; i < workerCount; i++ {
51       wg.Add(1)
52       go work(&wg)
53    }
54 
55    wg.Wait()
56    close(resultCh)
57 }
58 
59 func getResults(done chan bool) {
60    for result := range resultCh {
61       fmt.Printf("task id is %d, randomNum is %d, randomNumDigitsSum is %d\n",
62          result.task.id, result.task.randomNum, result.randomNumDigitsSum)
63    }
64    done <- true
65 }
66 
67 func main() {
68    startTime := time.Now()
69 
70    taskCount := 100
71    go createTasks(taskCount)
72 
73    workerCount := 10
74    go createWorkers(workerCount)
75 
76    done := make(chan bool)
77    go getResults(done)
78    <-done
79 
80    endTime := time.Now()
81    fmt.Printf("cost time: %fs\n", endTime.Sub(startTime).Seconds())
82 }

这里，工作池的任务是计算所输入数字的每一位的和。
taskCh和resultCh的通道容量都是10。
10个工作协程从taskCh里面不断获取值，把结果放入resultCh。
第一批10个任务进入taskCh后，每个工作协程获取后等待2s，第二批开始进入。
2s后，第二批就被每个工作协程获取到，第一批结果被打印出来。
直到结束，耗时大约20s。

运行结果

 1 ——————————等待2秒
 2 task id is 4, randomNum is 983, randomNumDigitsSum is 20
 3 task id is 2, randomNum is 636, randomNumDigitsSum is 15
 4 task id is 10, randomNum is 150, randomNumDigitsSum is 6
 5 task id is 7, randomNum is 520, randomNumDigitsSum is 7
 6 task id is 5, randomNum is 895, randomNumDigitsSum is 22
 7 task id is 9, randomNum is 904, randomNumDigitsSum is 13
 8 task id is 8, randomNum is 998, randomNumDigitsSum is 26
 9 task id is 3, randomNum is 407, randomNumDigitsSum is 11
10 task id is 1, randomNum is 878, randomNumDigitsSum is 23
11 task id is 6, randomNum is 735, randomNumDigitsSum is 15
12 ——————————等待2秒
13 task id is 12, randomNum is 538, randomNumDigitsSum is 16
14 task id is 11, randomNum is 212, randomNumDigitsSum is 5
15 task id is 14, randomNum is 362, randomNumDigitsSum is 11
16 task id is 15, randomNum is 436, randomNumDigitsSum is 13
17 task id is 13, randomNum is 750, randomNumDigitsSum is 12
18 task id is 17, randomNum is 630, randomNumDigitsSum is 9
19 task id is 16, randomNum is 215, randomNumDigitsSum is 8
20 task id is 20, randomNum is 914, randomNumDigitsSum is 14
21 task id is 19, randomNum is 20, randomNumDigitsSum is 2
22 task id is 18, randomNum is 506, randomNumDigitsSum is 11
23 ——————————等待2秒
24 一直到100...
25 cost time: 20.024336s

通过Go来处理每分钟达百万的数据请求：https://blog.csdn.net/tybaoerge/article/details/50392386

taskChPool里面的taskCh是非缓冲通道。
worker把自己的taskCh放入taskChPool，dispatch从taskChPool中取出taskCh并把task放入取出的taskCh中。
worker从自己的taskCh中取出任务并执行完成后，再次把自己的taskCh放入taskChPool，等待新的task。

 1 package main
 2 
 3 import (
 4    "fmt"
 5    "math/rand"
 6    "time"
 7 )
 8 
 9 type Task struct {
10    id        int
11    randomNum int
12 }
13 
14 type Worker struct {
15    taskCh chan Task
16 }
17 
18 var taskCh = make(chan Task, 10)
19 var taskChPool chan chan Task
20 
21 func createTasks(taskCount int) {
22    for i := 0; i < taskCount; i++ {
23       taskCh <- Task{i + 1, rand.Intn(999)}
24    }
25    close(taskCh)
26 }
27 
28 func getRandomNumDigitsSum(num int) int {
29    sum := 0
30    for num > 0 {
31       sum += num % 10
32       num /= 10
33    }
34    time.Sleep(2 * time.Second)
35    return sum
36 }
37 
38 func (w *Worker) work() {
39    for {
40       taskChPool <- w.taskCh
41       if task, ok := <-w.taskCh; ok {
42          fmt.Printf("task id is %d, randomNum is %d, randomNumDigitsSum is %d\n",
43             task.id, task.randomNum, getRandomNumDigitsSum(task.randomNum))
44       }
45    }
46 }
47 
48 func createWorkers(workerCount int) {
49    for i := 0; i < workerCount; i++ {
50       worker := Worker{make(chan Task)}
51       go worker.work()
52    }
53 }
54 
55 func dispatch() {
56    for task := range taskCh {
57       go func(task Task) {
58          taskCh := <-taskChPool
59          taskCh <- task
60       }(task)
61    }
62 }
63 
64 func main() {
65    taskCount := 100
66    go createTasks(taskCount)
67 
68    workerCount := 10
69    go createWorkers(workerCount)
70 
71    taskChPool = make(chan chan Task, workerCount)
72    go dispatch()
73 
74    time.Sleep(30 * time.Second)
75 }

两种方式对比：
请求是否会因处理请求过慢而阻塞。
第一个方案，会，如果处理请求过慢，那么会阻塞向taskCh里面放入task，相当于阻塞请求。
第二个方案，不会，如果处理请求过慢，那么把请求通过创建协程的方式来异步缓存起来，等worker有空了再执行，不是一直阻塞请求处理直到有worker空了。

数据传递（令牌）

经典地使用Channel进行任务编排的题：有四个 goroutine，编号为 1、2、3、4。每秒钟会有一个 goroutine 打印出它自己的编号，要求你编写一个程序，让输出的编号总是按照 1、2、3、4、1、2、3、4、……的顺序打印出来。
为了实现顺序的数据传递，定义一个令牌的变量，谁得到令牌，谁就可以打印一次自己的编号，同时将令牌传递给下一个 goroutine。

 1 package main
 2 
 3 import (
 4    "fmt"
 5    "time"
 6 )
 7 
 8 type Token struct{}
 9 
10 func newWorker(id int, ch chan Token, nextCh chan Token) {
11    for {
12       token := <-ch         // 取得令牌
13       fmt.Println(id + 1) // 打印id
14       time.Sleep(time.Second)
15       nextCh <- token
16    }
17 }
18 
19 func main() {
20    chs := []chan Token{make(chan Token), make(chan Token), make(chan Token), make(chan Token)}
21 
22    // 创建4个worker即启动4个协程分别去读1个channel
23    for i := 0; i < 4; i++ {
24       go newWorker(i, chs[i], chs[(i+1)%4])
25    }
26 
27    //首先把令牌交给第一个worker
28    chs[0] <- Token{}
29 
30    select {}
31 }

运行结果

定义一个令牌类型（Token）。
定义创建worker的函数（从自己的chan中读取令牌）。哪个goroutine取得了令牌，就可以打印出自己的编号。因为需要每秒打印一次数据，所以休眠1秒后，再把令牌交给它的下家。
在第24行启动每个worker的goroutine，并在第28行将令牌先交给第一个worker。

信号通知（程序优雅退出）

 1 func main() {
 2   go func() {
 3       ...... // 执行业务处理
 4   }()
 5 
 6   // 处理CTRL+C等中断信号
 7   termChan := make(chan os.Signal)
 8   signal.Notify(termChan, syscall.SIGINT, syscall.SIGTERM)
 9   <-termChan
10 
11   // 执行退出之前的清理动作
12   doCleanup()
13   
14   fmt.Println("优雅退出")
15 }

doCleanup可能是一个很耗时的操作，例如十几分钟才能完成。如果程序退出需要等待这么长时间，那么用户是不能接受的。所以，设置一个最长的等待时间。只要超过了这个时间，程序就不再等待，可以直接退出。所以，退出的时候分为两个阶段：closing代表程序退出，但是清理工作还没做；closed，代表清理工作已经做完。

 1 func main() {
 2     var closing = make(chan struct{})
 3     var closed = make(chan struct{})
 4 
 5     go func() {
 6         // 模拟业务处理
 7         for {
 8             select {
 9             case <-closing:
10                 return
11             default:
12                 // ....... 业务计算
13                 time.Sleep(100 * time.Millisecond)
14             }
15         }
16     }()
17 
18     // 处理CTRL+C等中断信号
19     termChan := make(chan os.Signal)
20     signal.Notify(termChan, syscall.SIGINT, syscall.SIGTERM)
21     <-termChan
22 
23       // 停止业务处理
24     close(closing)
25     // 执行退出之前的清理动作
26     go doCleanup(closed)
27 
28     select {
29     case <-closed:
30     case <-time.After(time.Second):
31         fmt.Println("清理超时，不等了")
32     }
33     fmt.Println("优雅退出")
34 }
35 
36 func doCleanup(closed chan struct{}) {
37     time.Sleep((time.Minute))
38     close(closed)
39 }

码农公寓

消息交流（生产者和消费者）

数据传递（令牌）

信号通知（程序优雅退出）

相关文章