当前位置： 首页 > article >正文

go当中的channel 无缓冲channel和缓冲channel的适用场景、结合select的使用

article 2024/11/17 22:40:45

Channel

Go channel就像Go并发模型中的“胶水”，它将诸多并发执行单元连接起来，或者正是因为有channel的存在，Go并发模型才能迸发出强大的表达能力。

无缓冲channel

无缓冲channel兼具通信和同步特性，在并发程序中应用颇为广泛。

可以通过不带有capacity参数的内置make函数创建一个可用的无缓冲channel：

c := make(chan T)

由于无缓冲channel的运行时层实现不带有缓冲区，因此对无缓冲channel的接收和发送操作是同步的。

一个无缓冲的channel动作发生和完成的时序如下：

  • 发送动作一定发生在接收动作完成之前；
  • 接收动作一定发生在发送动作完成之前。

这与Go官方“Go内存模型”一文中对channel通信的描述是一致的。正因如此，下面的代码可以保证main输出的变量a的值为"hello, world"：

var c = make(chan int)
var a string

func f() {
    a = "hello, world"
    <-c
}

func main() {
    go f()
    c <- 5
    println(a)
}

因为函数f中的channel接收动作发生在主goroutine对channel发送动作完成之前，而a = “hello, world"语句又发生在channel接收动作之前，因此主goroutine在channel发送操作完成后看到的变量a的值一定是"hello, world”，而不是空字符串。

用作信号传递

1）一对一通知信号

无缓冲channel常被用于在两个goroutine之间一对一地传递通知信号，

(2)一对多通知信号

无缓冲channel还被用来实现一对多的信号通知机制

关闭一个无缓冲channel会让所有阻塞在该channel上的接收操作返回，从而实现一种一对多的广播机制。该一对多的信号通知机制还常用于通知一组worker goroutine退出：

// 通知其他goroutine工作线程已完成
type signal struct{}

func worker(i int) {
	fmt.Printf("worker %v is working\n", i)
	time.Sleep(5 * time.Second)
	fmt.Printf("worker %v : works done\n", i)
}

// spawnGroup 是一个用于生成一组工作线程的函数
// 参数:
//   - f: 工作函数，每个工作线程都会执行该函数
//   - num: 工作线程的数量
//   - groupSignal: 用于控制工作线程启动和停止的信号通道
//
// 返回值:
//   - <-chan signal: 用于接收所有工作线程完成的信号通道
func spawnGroup(f func(i int), num int, groupSignal <-chan signal) <-chan signal {
	c := make(chan signal)
	var wg sync.WaitGroup
	for i := 0; i < num; i++ {
		wg.Add(1)
		go func(i int) {
			<-groupSignal //阻塞，等待启动信号
			fmt.Printf("worker %v: start to work\n", i)
			f(i)
			wg.Done() //工作完成，减少WaitGroup的计数
		}(i + 1)
	}

	go func() {
		wg.Wait()
		c <- signal(struct{}{}) //发送信号通知
	}()
	return c
}

func main() {
	fmt.Println("start a group of workers...")
	groupSignal := make(chan signal)
	c := spawnGroup(worker, 5, groupSignal)
	time.Sleep(10 * time.Second)
	fmt.Println("the group of workers start to work...")
	// 关闭工作组信号通道，通知所有工作线程开始工作
	close(groupSignal)
	<-c
	fmt.Println("the group of workers work done!")
}

最后结果：

Untitled

用于替代锁机制

由于无缓冲channel具有同步特性，因此可以在某些场合替代锁，让程序更加清晰，可读性增强；以下给出基于共享内存+锁模式的goroutine安全的计数器：

type counter struct {
	c chan int
	i int
}

var cter counter

func InitCounter() {
	cter = counter{
		c: make(chan int),
	}
	// 增加计数器的动作相当于一次接收动作
	go func() {
		for {
			cter.i++
			cter.c <- cter.i
		}
	}()
	fmt.Println("counter init ok")
}

func Increase() int {
	return <-cter.c
}

func init() {
	InitCounter()
}

func main() {
	for i := 0; i < 10; i++ {
		go func(i int) {
			v := Increase()
			fmt.Printf("goroutine-%d: current counter value is %d\n", i, v)
		}(i)
	}
	time.Sleep(5 * time.Second)
}

此代码通过无缓冲channel的同步阻塞特性实现计数器的控制

也符合“不要通过共享内存来通信，而应该通过通信来共享内存”的原则

缓冲channel

带缓冲channel可以通过带有capacity参数的内置make函数创建

c := make(chan T, capacity)

接收操作在缓冲区非空的情况下是异步的（发送或接收无须阻塞等待）

用作消息队列

channel的原生特性与我们认知中的消息队列十分相似，包括goroutine安全、有fifo（first-in, first out）保证等。异步收发的带缓冲channel更适合用作消息队列，并且带缓冲channel在数据收发性能上要明显好于无缓冲channel

用作计数信号量 counting semaphore

带缓冲channel当前数据个数代表的是同时处于活跃状态的goroutine数量，capacity则代表同时处于活跃状态的最大数量。以下是一个例子：

// 同一时间最多3个活跃状态
var active = make(chan struct{}, 3)
var jobs = make(chan int, 10)

func main() {
	go func() {
		for i := 0; i < 8; i++ {
			jobs <- i + 1
		}
		close(jobs)
	}()
	var wg sync.WaitGroup
	for j := range jobs {
		wg.Add(1)
		go func(j int) {
			active <- struct{}{}
			log.Printf("handle job: %v\n", j)
			time.Sleep(2 * time.Second)
			<-active
			wg.Done()
		}(j)
	}
	wg.Wait()
}

结果：

Untitled

可以发现同一时间处于处理状态的job最多为3个

len(channel)的应用

如果s是chan T类型，那么len(s)针对channel的类型不同，有如下两种语义：

  • 当s为无缓冲channel时，len(s)总是返回0；
  • 当s为带缓冲channel时，len(s)返回当前channel s中尚未被读取的元素个数。

但是单纯依靠if语句来判断channel元素状态并不可靠，因为在并发状态下不能保证后续对channel进行收发时channel状态不变：

Untitled

oroutine1在使用len(channel)判空后，便尝试从channel中接收数据。但在其真正从channel中读数据前，goroutine2已经将数据读了出去，goroutine1后面的读取将阻塞在channel上，导致后面逻辑失效。因此，为了不阻塞在channel上，常见的方法是将判空与读取放在一个事务中，将判满与写入放在一个事务中，而这类事务我们可以通过select实现。来看下面的示例：

func producer(c chan<- int) {
	var i int = 1
	for {
		time.Sleep(2 * time.Second)
		ok := trySend(c, i)
		if ok {
			fmt.Printf("[producer]: send [%d] to channel\n", i)
			i++
			continue
		}
		fmt.Printf("[producer]: try send [%d], but channel is full\n", i)
	}
}

func tryRecv(c <-chan int) (int, bool) {
	select {
	case i := <-c:
		return i, true
	default:
		return 0, false
	}
}

func trySend(c chan<- int, i int) bool {
	select {
	case c <- i:
		return true
	default:
		return false
	}
}

func consumer(c <-chan int) {
	for {
		i, ok := tryRecv(c)
		if !ok {
			fmt.Println("[consumer]: try to recv from channel, but the channel is empty")
			time.Sleep(1 * time.Second)
			continue
		}
		fmt.Printf("[consumer]: recv [%d] from channel\n", i)
		if i >= 3 {
			fmt.Println("[consumer]: exit")
			return
		}
	}
}

func main() {
	c := make(chan int, 3)
	go producer(c)
	go consumer(c)

	select {} // 仅用于演示，临时用来阻塞主goroutine
}

结果：

Untitled

这种方法的缺点就在于改变了channel的状态

想在不改变channel状态的前提下单纯地侦测channel的状态，又不会因channel满或空阻塞在channel上。但很遗憾，目前没有一种方法既可以实现这样的功能又适用于所有场合。在特定的场景下，可以用len(channel)来实现。比如图34-2中的这两种场景。

在图34-2中，a是一个多发送单接收的场景，即有多个发送者，但有且只有一个接收者。在这样的场景下，我们可以在接收者goroutine中根据len(channel)是否大于0来判断channel中是否有数据需要接收。

b是一个多接收单发送的场景，即有多个接收者，但有且只有一个发送者。在这样的场景下，我们可以在发送goroutine中根据len(channel)是否小于cap(channel)来判断是否可以执行向channel的发送操作。

Untitled

nil channel的妙用

没有初始化的channel（nil channel）进行读写操作将会发生阻塞

func main() {
	var c chan int
	<-c
}

结果：

Untitled

main goroutine被阻塞在channel上，导致Go运行时认为出现deadlock状态并抛出panic。

但nil channel并非一无是处。来看一个例子：

func main() {
    c1, c2 := make(chan int), make(chan int)
    go func() {
        time.Sleep(time.Second * 5)
        c1 <- 5
        close(c1)
    }()
    
    go func() {
        time.Sleep(time.Second * 7)
        c2 <- 7
        close(c2)
    }()
    
    for {
        select {
        case x, ok := <-c1:
	// 对于一个nil channel执行获取操作，该操作会被堵塞，因此可以显示设置
            if !ok {
                c1 = nil
            } else {
                fmt.Println(x)
            }
        case x, ok := <-c2:
                if !ok {
                    c2 = nil
                } else {
                    fmt.Println(x)
                }
        }
        if c1 == nil && c2 == nil {
            break
        }
    }
    fmt.Println("program end")
}

与select结合

避免阻塞

default的使用通常是在没得选的情况下，因此也有一种可以避免堵塞的特性

func sendTime(c interface{}, seq uintptr) {
    // 无阻塞地向c发送当前时间
    // ...
    select {
        case c.(chan Time) <- Now():
        default:
    }
}

实现超时机制

通过超时事件，既可以避免陷入无尽的等待也可以做一些异常处理工作：

func worker() {
	select {
	case <-c:
		//...
	case <-time.After(30*time.Second):
		return
	}
}

timer实质上是由go运行时自动维护的，而不是操作系统的定时器资源：

Untitled

go通过名为timerproc的函数，维护了一个“最小堆”。该goroutine会被定期唤醒并读取堆顶的timer对象，执行该timer对象对应的函数（向timer.C中发送一条数据，触发定时器），执行完毕后就会从最小堆中移除该timer对象。

所以我们在使用timer的时候应该即使调用timer的Stop方法从最小堆中删除尚未到达过期时间的timer对象。

实现心跳机制

这种机制可以使我们在监听的同时执行一些周期性任务，比如下面这段代码：

func worker() {
    heartbeat := time.NewTicker(30 * time.Second)
    defer heartbeat.Stop()
    for {
        select {
        case <-c:
            // ... 处理业务逻辑
        case <- heartbeat.C: //记得调用方法停止运作
            //... 处理心跳
        }
    }
}

http://www.kler.cn/a/147393.html

相关文章：

  • ODC 如何精确呈现SQL耗时 | OceanBase 开发者工具解析
  • 华为Ensp模拟器配置RIP路由协议
  • 控制器ThinkPHP6
  • 推荐15个2024最新精选wordpress模板
  • 使用Java绘制图片边框，解决微信小程序map组件中marker与label层级关系问题，label增加外边框后显示不能置与marker上面
  • 15分钟学 Go 第 56 天：架构设计基本原则
  • window关于下载anaconda 2023年以后的版本，jupyter notebook闪退，没有内核的问题
  • Golang Proxy Protocol详解
  • vue3中如何使用TypeScript？
  • HuggingFace学习笔记--metrics和pipeline的使用
  • SpringBoot : ch10 整合Elasticsearch
  • 爬虫必学：Java创建代理ip池详细教程
  • flink的java.lang.IllegalStateException: Buffer pool is destroyed 异常
  • 大坝安全监测的内容及作用
  • 给定序列a，选k个数排成一排，从左往右扫，如果当前数小于上一个数，那么当前数变成上一个数，然后形成最后的序列，问形成的序列有多少种
  • 虚幻学习笔记2—点击场景3D物体的两种处理方式
  • 电子学会C/C++编程等级考试2022年09月（二级）真题解析
  • 项目中如何配置数据可视化展现
  • 【Java】IDEA 基本操作
  • Java 简易版王者荣耀
  • rust-flexi_logger
  • 40.0/jdbc/Java数据连接/jar包运用增删改
  • iOS强引用引起的内存泄漏
  • 基于单片机设计的超声波测距仪(采用HC-SR04模块)
  • 【限时免费】20天拿下华为OD笔试之【双指针】2023Q1A-两数之和绝对值最小【欧弟算法】全网注释最详细分类最全的华为OD真题题解
  • 二十章多线程