Golang实现带优先级的select

作者:jxwu 时间:2024-04-26 17:36:41 

背景

在 Golang 里面,我们经常使用 channel 进行协程之间的通信。这里有一个经典的场景,也就是生产者消费者模式,生产者协程不断地往 Channel 里面塞元素,而消费者协程不断地消费这些元素。

Golang实现带优先级的select

写成代码就是如下:

package main

import (
"fmt"
"sync"
"time"
)

func main() {
ch := make(chan int, 10)
var wg sync.WaitGroup
wg.Add(2)
go producer(ch, &wg)
go consumer(ch, &wg)
wg.Wait()
}

// 生产者
func producer(ch chan int, wg *sync.WaitGroup) {
defer wg.Done()
i := 0
for {
select {
case ch <- i:
default:
// 丢弃
log.Println("discard")
}
i++
time.Sleep(time.Second)
}
}

// 消费者
func consumer(ch chan int, wg *sync.WaitGroup) {
defer wg.Done()
consume := func(i int) {
fmt.Println(i)
time.Sleep(time.Millisecond * 700)
}
for {
i := <-ch
consume(i) // 消费元素
}
}

生产者不断产生元素,消费者消费元素。生产者不会等待消费者消费完毕(不然可能影响其他任务),如果 channel 已经满了,也就是说明消费者消费不过来,生产者就会丢弃这个任务。

生产者平均一秒生成1个,消费者0.7秒消费一个。正常情况下消费者是消费得过来的,然而很多时候消费者协程还需要做一些定时任务,比如一些定时清理工作。假如这个清理工作每2秒触发一次,清理时间一般需要1.5秒,也就是如果每次都做每一秒有0.75秒会被清理工作占有了,但是它不是一定要非常及时的,可以等空闲时再进行。 如下代码:

// 消费者
func consumer(ch chan int, wg *sync.WaitGroup) {
defer wg.Done()
t := time.NewTicker(time.Second * 2)
consume := func(i int) {
fmt.Println(i)
time.Sleep(time.Millisecond * 700)
}
clear := func() {
fmt.Println("clear")
time.Sleep(time.Millisecond * 1500)
}
for {
select {
case i := <-ch:
consume(i) // 消费元素:
case <-t.C:
clear() // 清理
}
}
}

运行程序到第15秒的时候,生产者发现 channel满了,于是开始丢包:

0
1
clear
2
3
4
5
6
clear
7
clear
8
clear
9
clear
clear
10
clear
11
12
13
14
clear
15
clear
clear
discard
16
clear
discard
discard

解决方案

既然清理任务的优先级并不高,那么它就不应该阻塞消费元素流程,而是应该在空闲时才去执行。由于 Golang 里面,如果 select 两个 case 都同时满足,会随机选一个执行,因此第一想到的可能会使用如下代码实现优先级case:

// 消费者
func consumer(ch chan int, wg *sync.WaitGroup) {
defer wg.Done()
t := time.NewTicker(time.Second * 2)
consume := func(i int) {
fmt.Println(i)
time.Sleep(time.Millisecond * 700)
}
clear := func() {
fmt.Println("clear")
time.Sleep(time.Millisecond * 1500)
}
for {
select {
case i := <-ch:
consume(i) // 消费元素
continue   // 可能还有元素,不走清理逻辑
default:
}

// 没有元素才走清理逻辑
select {
case <-t.C:
clear() // 清理
default:
}
}
}

如果运行这个程序,可以发现它能够满足优先级的需求,先消费元素,空闲时再执行清理任务。

然而,在没有元素可以消费,也没有清理任务可以执行的时候,这里的for将会不断地循环,浪费CPU资源。

其实,可以使用下面的方法实现优先级case,它能够在没有元素就绪的时候阻塞在 select,而不是不断循环:

// 消费者
func consumer(ch chan int, wg *sync.WaitGroup) {
defer wg.Done()
t := time.NewTicker(time.Second * 2)
consume := func(i int) {
fmt.Println(i)
time.Sleep(time.Millisecond * 700)
}
clear := func() {
fmt.Println("clear")
time.Sleep(time.Millisecond * 1500)
}
for {
select {
case i := <-ch:
consume(i) // 消费元素
case <-t.C:
priority:
for { // 清理前先把元素消费完
select {
case i := <-ch:
consume(i) // 消费元素
default:
break priority // 注:这里会跳过这个循环,而不是再次执行
}
}
clear() // 清理
}
}
}

这里的关键是在触发清理case的时候,先去把channel里面的元素消费完,再进行清理,从而保证能够留下足够的channel缓冲区给生产者放置生产的元素。

一个封装

上面那段优先级case代码其实挺常用的,但是几乎都是模板代码,特别是需要在两个地方写consume(i),因此我们可以封装一下这段代码,方便使用,减少出错:

// 优先级select ch1 的任务先执行完毕后才会执行 ch2 里面的任务
func PrioritySelect[T1, T2 any](ch1 <-chan T1, f1 func(T1), ch2 <-chan T2, f2 func(T2)) {
for {
select {
case a := <-ch1:
f1(a)
case b := <-ch2:
priority:
for {
select {
case a := <-ch1:
f1(a)
default:
break priority
}
}
f2(b)
}
}
}

这样,我们的消费者代码就可以简化为:

// 消费者
func consumer(ch chan int, wg *sync.WaitGroup) {
defer wg.Done()
t := time.NewTicker(time.Second * 2)
consume := func(i int) {
fmt.Println(i)
time.Sleep(time.Millisecond * 700)
}
clear := func(time.Time) {
fmt.Println("clear")
time.Sleep(time.Millisecond * 1500)
}
PrioritySelect(ch, consume, t.C, clear)
}

来源:https://juejin.cn/post/7217082232938053693

标签:Golang,优先级,select
0
投稿

猜你喜欢

  • 让sql2005运行在独立用户下出现 WMI 提供程序错误的解决方式

    2024-01-13 13:12:50
  • 渗透和改变 非洲web 2.0网站Logo

    2008-01-22 13:31:00
  • 利用JS提交表单的几种方法和验证(必看篇)

    2023-08-17 16:43:22
  • PDO::prepare讲解

    2023-06-06 06:15:28
  • JavaScript中filter的用法实例分析

    2024-04-10 10:59:42
  • 八种获取当前日期的js代码

    2007-09-07 10:26:00
  • vue-cli-service build 环境设置方式

    2024-05-25 15:17:08
  • pandas和spark dataframe互相转换实例详解

    2022-12-12 20:26:38
  • Java 数据库连接池c3p0 介绍

    2024-01-22 02:44:24
  • python 格式化输出百分号的方法

    2023-07-17 08:23:27
  • Python导入不同文件夹中文件的方法详解

    2022-01-22 09:30:08
  • 详解django2中关于时间处理策略

    2021-09-09 23:13:09
  • Autopep8的使用(python自动编排工具)

    2022-07-04 11:10:47
  • 使用Python生成随机密码的示例分享

    2023-05-23 11:32:10
  • 源码分析系列之json_encode()如何转化一个对象

    2023-07-15 05:25:54
  • 使用PyInstaller将Python程序文件转换为可执行程序文件

    2023-03-28 11:35:34
  • Python爬虫之线程池的使用

    2021-04-08 20:01:01
  • pytorch_detach 切断网络反传方式

    2022-09-25 21:10:50
  • 如何彻底解决python NameError:name '__file__' is not defined

    2023-01-29 03:57:12
  • python 工具类之Queue组件详解用法

    2023-08-05 23:59:10
  • asp之家 网络编程 m.aspxhome.com