本篇内容介绍了“Go中的并发方法实例代码分析”的有关知识,在实际案例的操作过程中,不少人都会遇到这样的困境,接下来就让小编带领大家学习一下如何处理这些情况吧!希望大家仔细阅读,能够学有所成!
代码很简单——单个通道,单个goroutine,单次写入,单次读取。
package main
func main() {
// 创建一个int类型的通道
ch := make(chan int)
// 开启一个匿名 goroutine
go func() {
// 向通道发送数字42
ch <- 42
}()
// 从通道中读取
<-ch
}
转到交互式 WebGL 动画
蓝色线代表随时间运行的goroutine. 连接‘main’和‘go #19’的蓝色细线用来标记goroutine的开始和结束同时展示了父子关系,最后,红线代表发送/接收动作. 虽然这是两个独立的动作,我还是尝试用“从 A 发送到 B”的动画将他们表示成一个动作. goroutine 名称中的“#19” 是 goroutine 真实的内部ID, 其获取方法参考了 Scott Mansfield 的 “Goroutine IDs” 这篇文章。
实际上,你可以通过以下方法构建一个简单的计时器——创建一个通道, 开启一个 goroutine 让其在指定的时间间隔后向通道中写入数据,然后将这个通道返回给调用者。于是调用函数就会在读取通道时阻塞,直到之前设定的时间间隔过去。接下来我们调用24次计时器然后尝试具象化调用过程。
package main
import "time"
func timer(d time.Duration) <-chan int {
c := make(chan int)
go func() {
time.Sleep(d)
c <- 1
}()
return c
}
func main() {
for i := 0; i < 24; i++ {
c := timer(1 * time.Second)
<-c
}
}
转到交互式 WebGL 动画
很整洁,对吗? 我们继续。
这个并发例子取自谷歌员工 Sameer Ajmani “Advanced Go Concurrency Patterns” 演讲。当然,这个模式不算非常高级,但是对于那些只熟悉Go的并发机制的人来说它看起来可能非常新鲜有趣。
这里我们用一个通道代表乒乓球台. 一个整型变量代表球, 然后用两个goroutine代表玩家,玩家通过增加整型变量的值(点击计数器)模拟击球动作。
package main
import "time"
func main() {
var Ball int
table := make(chan int)
go player(table)
go player(table)
table <- Ball
time.Sleep(1 * time.Second)
<-table
}
func player(table chan int) {
for {
ball := <-table
ball++
time.Sleep(100 * time.Millisecond)
table <- ball
}
}
转到交互式 WebGL 动画
这里我建议你点击 链接 进入交互式 WebGL 动画操作一下. 你可以放慢或者加速动画,从不同的角度观察。
现在,我们添加三个玩家看看。
go player(table)
go player(table)
go player(table)
转到交互式 WebGL 动画 我们可以看到每个玩家都按照次序轮流操作,你可能会想为什么会这样。为什么多个玩家(goroutine)会按照严格的顺序接到“球”呢。
答案是 Go 运行时环境维护了一个 接收者 FIFO 队列 (存储需要从某一通道上接收数据的goroutine),在我们的例子里,每个玩家在刚发出球后就做好了接球准备。我们来看一下更复杂的情况,加入100个玩家。
for i := 0; i < 100; i++ {
go player(table)
}
转到交互式 WebGL 动画
先进先出顺序很明显了,是吧? 我们可以创建一百万个goroutine,因为它们很轻量,但是对于实现我们的目的来说没有必要。我们来想想其他可以玩的。 例如, 常见的消息传递模式。
并发世界中流行的模式之一是所谓的 fan-in 模式。这与 fan-out 模式相反,稍后我们将介绍。简而言之,fan-in 是一项功能,可以从多个输入中读取数据并将其全部多路复用到单个通道中。
举例来说:
package main
import (
"fmt"
"time"
)
func producer(ch chan int, d time.Duration) {
var i int
for {
ch <- i
i++
time.Sleep(d)
}
}
func reader(out chan int) {
for x := range out {
fmt.Println(x)
}
}
func main() {
ch := make(chan int)
out := make(chan int)
go producer(ch, 100*time.Millisecond)
go producer(ch, 250*time.Millisecond)
go reader(out)
for i := range ch {
out <- i
}
}
Go to interactive WebGL animation
如我们所见,第一个 producer 每100毫秒生成一次值,第二个每250毫秒生成一次值,但是 reader 会立即从这两个生产者那里接受值。实际上,多路复用发生在 main 的range循环中。
与 fan-in 相反的模式是 fan-out 或者worker 模式。多个 goroutine 可以从单个通道读取,从而在CPU内核之间分配大量的工作量,因此是 worker 的名称。在Go中,此模式易于实现-只需以通道为参数启动多个goroutine,然后将值发送至该通道-Go运行时会自动地进行分配和复用 :)
package main
import (
"fmt"
"sync"
"time"
)
func worker(tasksCh <-chan int, wg *sync.WaitGroup) {
defer wg.Done()
for {
task, ok := <-tasksCh
if !ok {
return
}
d := time.Duration(task) * time.Millisecond
time.Sleep(d)
fmt.Println("processing task", task)
}
}
func pool(wg *sync.WaitGroup, workers, tasks int) {
tasksCh := make(chan int)
for i := 0; i < workers; i++ {
go worker(tasksCh, wg)
}
for i := 0; i < tasks; i++ {
tasksCh <- i
}
close(tasksCh)
}
func main() {
var wg sync.WaitGroup
wg.Add(36)
go pool(&wg, 36, 50)
wg.Wait()
}
这里值得一提的是:并行性。如您所见,所有goroutine并行’运行‘,等待通道给予它们’工作‘。鉴于上面的动画,很容易发现goroutine几乎立即接连地收到它们的工作。不幸的是,该动画在goroutine确实在处理工作还是仅仅是在等待输入的地方没有用颜色显示出来,但是此动画是在GOMAXPROCS=4的情况下录制的,因此只有4个goroutine有效地并行运行。我们将很快讨论这个主题。
现在,让我们做一些更复杂的事情,并启动一些有自己workers(subworkers)的workers。
package main
import (
"fmt"
"sync"
"time"
)
const (
WORKERS = 5
SUBWORKERS = 3
TASKS = 20
SUBTASKS = 10
)
func subworker(subtasks chan int) {
for {
task, ok := <-subtasks
if !ok {
return
}
time.Sleep(time.Duration(task) * time.Millisecond)
fmt.Println(task)
}
}
func worker(tasks <-chan int, wg *sync.WaitGroup) {
defer wg.Done()
for {
task, ok := <-tasks
if !ok {
return
}
subtasks := make(chan int)
for i := 0; i < SUBWORKERS; i++ {
go subworker(subtasks)
}
for i := 0; i < SUBTASKS; i++ {
task1 := task * i
subtasks <- task1
}
close(subtasks)
}
}
func main() {
var wg sync.WaitGroup
wg.Add(WORKERS)
tasks := make(chan int)
for i := 0; i < WORKERS; i++ {
go worker(tasks, &wg)
}
for i := 0; i < TASKS; i++ {
tasks <- i
}
close(tasks)
wg.Wait()
}
Go to interactive WebGL animation 很好。当然,我们可以将worker和subworker的数量设置为更高的值,但是我试图使动画清晰易懂。
更酷的 fan-out 模式确实存在,例如动态数量的worker/subworker,通过通道发送通道,但是 fan-out 的想法现在应该很清楚了。
下一个常见的模式类似于扇出,但是会在很短的时间内生成goroutine,只是为了完成某些任务。它通常用于实现服务器-创建侦听器,循环运行accept()并为每个接受的连接启动goroutine。它非常具有表现力,可以实现尽可能简单的服务器处理程序。看一个简单的例子:
package main
import "net"
func handler(c net.Conn) {
c.Write([]byte("ok"))
c.Close()
}
func main() {
l, err := net.Listen("tcp", ":5000")
if err != nil {
panic(err)
}
for {
c, err := l.Accept()
if err != nil {
continue
}
go handler(c)
}
}
Go to 交互式WebGL动画
这不是很有趣-似乎并发方面没有发生任何事情。当然,在引擎盖下有很多复杂性,这是我们特意隐藏的。 “简单性很复杂”.
但是,让我们回到并发性并向我们的服务器添加一些交互。假设每个处理程序都希望异步写入记录器。在我们的示例中,记录器本身是一个单独的goroutine
,它可以完成此任务。
package main
import (
"fmt"
"net"
"time"
)
func handler(c net.Conn, ch chan string) {
ch <- c.RemoteAddr().String()
c.Write([]byte("ok"))
c.Close()
}
func logger(ch chan string) {
for {
fmt.Println(<-ch)
}
}
func server(l net.Listener, ch chan string) {
for {
c, err := l.Accept()
if err != nil {
continue
}
go handler(c, ch)
}
}
func main() {
l, err := net.Listen("tcp", ":5000")
if err != nil {
panic(err)
}
ch := make(chan string)
go logger(ch)
go server(l, ch)
time.Sleep(10 * time.Second)
}
Go to 交互式WebGL动画
不是吗?但是很容易看到,如果请求数量增加并且日志记录操作花费一些时间(例如,准备和编码数据),我们的* logger * goroutine很快就会成为瓶颈。我们可以使用一个已知的扇出模式。我们开始做吧。
带工作程序的服务器示例是记录器的高级版本。它不仅可以完成一些工作,而且还可以通过* results *通道将其工作结果发送回池中。没什么大不了的,但是它将我们的记录器示例扩展到了更实际的示例。
让我们看一下代码和动画:
package main
import (
"net"
"time"
)
func handler(c net.Conn, ch chan string) {
addr := c.RemoteAddr().String()
ch <- addr
time.Sleep(100 * time.Millisecond)
c.Write([]byte("ok"))
c.Close()
}
func logger(wch chan int, results chan int) {
for {
data := <-wch
data++
results <- data
}
}
func parse(results chan int) {
for {
<-results
}
}
func pool(ch chan string, n int) {
wch := make(chan int)
results := make(chan int)
for i := 0; i < n; i++ {
go logger(wch, results)
}
go parse(results)
for {
addr := <-ch
l := len(addr)
wch <- l
}
}
func server(l net.Listener, ch chan string) {
for {
c, err := l.Accept()
if err != nil {
continue
}
go handler(c, ch)
}
}
func main() {
l, err := net.Listen("tcp", ":5000")
if err != nil {
panic(err)
}
ch := make(chan string)
go pool(ch, 4)
go server(l, ch)
time.Sleep(10 * time.Second)
}
Go to 交互式WebGL动画 我们在4个goroutine之间分配了工作,有效地提高了记录器的吞吐量,但是从此动画中,我们可以看到记录器仍然可能是问题的根源。成千上万的连接在分配之前会汇聚在一个通道中,这可能导致记录器再次成为瓶颈。但是,当然,它会在更高的负载下发生。
足够的扇入/扇出乐趣。让我们看看更复杂的并发算法。我最喜欢的例子之一是Concurrent Prime Sieve,可以在[Go Concurrency Patterns]对话中找到。素数筛,或[Eratosthenes筛)是一种古老的算法,用于查找达到给定限制的素数。它通过按顺序消除所有质数的倍数来工作。天真的算法并不是真正有效的算法,尤其是在多核计算机上。
该算法的并发变体使用goroutine过滤数字-每个发现的素数一个goroutine,以及用于将数字从生成器发送到过滤器的通道。找到质数后,它将通过通道发送到* main *以进行输出。当然,该算法也不是很有效,特别是如果您想找到大质数并寻找最低的Big O复杂度,但是我发现它非常优雅。
// 并发的主筛
package main
import "fmt"
// 将序列2、3、4,...发送到频道“ ch”。
func Generate(ch chan<- int) {
for i := 2; ; i++ {
ch <- i // Send 'i' to channel 'ch'.
}
}
//将值从通道“ in”复制到通道“ out”,
//删除可被“素数”整除的那些。
func Filter(in <-chan int, out chan<- int, prime int) {
for {
i := <-in // Receive value from 'in'.
if i%prime != 0 {
out <- i // Send 'i' to 'out'.
}
}
}
//主筛:菊花链过滤器过程。
func main() {
ch := make(chan int) // Create a new channel.
go Generate(ch) // Launch Generate goroutine.
for i := 0; i < 10; i++ {
prime := <-ch
fmt.Println(prime)
ch2 := make(chan int)
go Filter(ch, ch2, prime)
ch = ch2
}
}
转到交互式WebGL动画
,请以交互模式随意播放此动画。我喜欢它的说明性-它确实可以帮助您更好地理解该算法。 * generate * goroutine发出从2开始的每个整数,每个新的goroutine仅过滤特定的质数倍数-2、3、5、7 …,将第一个找到的质数发送给* main *。如果旋转它从顶部看,您会看到从goroutine发送到main的所有数字都是质数。漂亮的算法,尤其是在3D中。
现在,让我们回到我们的工作人员示例。还记得我告诉过它以GOMAXPROCS = 4运行吗?那是因为所有这些动画都不是艺术品,它们是真实程序的真实痕迹。
让我们回顾一下GOMAXPROCS是什么。
GOMAXPROCS设置可以同时执行的最大CPU数量。
当然,CPU是指逻辑CPU。我修改了一些示例,以使他们真正地工作(而不仅仅是睡觉)并使用实际的CPU时间。然后,我运行了代码,没有进行任何修改,只是设置了不同的GOMAXPROCS值。 Linux机顶盒有2个CPU,每个CPU具有12个内核,因此有24个内核。
因此,第一次运行演示了该程序在1个内核上运行,而第二次-使用了所有24个内核的功能。
WebGL动画-1| WebGL动画-24GOMAXPROCS1
这些动画中的时间速度是不同的(我希望所有动画都适合同一时间/ height),因此区别很明显。当GOMAXPROCS = 1时,下一个工作人员只有在上一个工作完成后才能开始实际工作。在GOMAXPROCS = 24的情况下,加速非常大,而复用的开销可以忽略不计。
不过,重要的是要了解,增加GOMAXPROCS并不总是可以提高性能,在某些情况下实际上会使它变得更糟。
我们可以从Go中的并发时间中证明什么呢?我想到的一件事情是goroutine泄漏。例如,如果您启动goroutine,但超出范围,可能会发生泄漏。或者,您只是忘记添加结束条件,而运行了for{}循环。
第一次在代码中遇到goroutine泄漏时,我的脑海中出现了可怕的图像,并且在下个周末我写了 expvarmon。现在,我可以使用WebGL可视化该恐怖图像。
看一看:
仅仅是看到此,我都会感到痛苦:) 所有这些行都浪费了资源,并且是您程序的定时炸弹。
我要说明的最后一件事是并行性与并发性之间的区别。这个话题涵盖了 很多 ,Rob Pike在这个话题上做了一个精彩的演讲。确实是#必须观看的视频之一。
简而言之,
并行是简单的并行运行事物。
并发是一种构造程序的方法。
因此,并发程序可能是并行的,也可能不是并行的,这些概念在某种程度上是正交的。我们在演示 GOMAXPROCS 设置效果时已经看到了这一点。
我可以重复所有这些链接的文章和谈话,但是一张图片相当于说了一千个字。我在这里能做的是可视化这个差异。因此,这是并行。许多事情并行运行。
转到交互式WebGL动画
这也是并行性:
转到交互式WebGL动画
但这是并发的:
还有这个:
这也是并发的:
为了创建这些动画,我编写了两个程序:gotracer 和 gothree.js 库。首先,gotracer执行以下操作:
解析Go程序的AST树(Abstract Syntax Tree,抽象语法树),并在与并发相关的事件上插入带有输出的特殊命令-启动/停止goroutine,创建通道,向/从通道发送/接收。
运行生成的程序
分析此特殊输出,并生成带有事件和时间戳描述的JSON。
生成的JSON示例:
“Go中的并发方法实例代码分析”的内容就介绍到这里了,感谢大家的阅读。如果想了解更多行业相关的知识可以关注亿速云网站,小编将为大家输出更多高质量的实用文章!
亿速云「云服务器」,即开即用、新一代英特尔至强铂金CPU、三副本存储NVMe SSD云盘,价格低至29元/月。点击查看>>
免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。