Golang-简洁的并发

转载原文:http://www.yankay.com/go-clear-concurreny/

多核处理器越来越普及。有没有一种简单的办法,能够让我们写的软件释放多核的威力?是有的。随着Golang, Erlang, Scala等为并发设计的程序语言的兴起,新的并发模式逐渐清晰。正如过程式编程和面向对象一样,一个好的编程模式有一个极其简洁的内核,还有在此之上丰富的外延。可以解决现实世界中各种各样的问题。本文以GO语言为例,解释其中内核、外延。

并发模式之内核

这种并发模式的内核只需要 协程通道 就够了。协程负责执行代码,通道负责在协程之间传递事件。

不久前,并发编程是个非常困难的事。要想编写一个良好的并发程序,我们不得不了解线程,锁,semaphore,barrier甚至CPU更新高速缓存的方式,而且他们个个都有怪脾气,处处是陷阱。笔者除非万不得以,决不会自己操作这些底层并发元素。一个简洁的并发模式不需要这些复杂的底层元素,协程和通道就够了。

协程是轻量级的线程。在过程式编程中,当调用一个过程的时候,需要等待其执行完才返回。而调用一个协程的时候,不需要等待其执行完,会立即返回。协程十分轻量,Go语言可以在一个进程中执行有数以十万计的协程,依旧保持高性能。而对于普通的平台,一个进程有数千个线程,其CPU会忙于上下文切换,性能急剧下降。随意创建线程可不是一个好主意,但是我们可以大量使用的协程。

通道是协程之间的数据传输通道。通道可以在众多的协程之间传递数据,具体可以值也可以是个引用。通道有两种使用方式。

  • 协程可以试图向通道放入数据,如果通道满了,会挂起协程,直到通道可以为他放入数据为止。
  • 协程可以试图向通道索取数据,如果通道没有数据,会挂起协程,直到通道返回数据为止。

如此,通道就可以在传递数据的同时,控制协程的运行。有点像事件驱动,也有点像阻塞队列。

这两个概念非常的简单,各个语言平台都会有相应的实现。在Java和C上也各有库可以实现两者。

Golang

Erlang

Scala(Actor)

协程

goroutines

process

actor

消息队列

channel

mailbox

channel

只要有协程和通道,就可以优雅的解决并发的问题。不必使用其他和并发有关的概念。那如何用这两把利刃解决各式各样的实际问题呢?

并发模式之外延

协程相较于线程,可以大量创建。打开这扇门,我们拓展出新的用法,可以做生成器,可以让函数返回“服务”,可以让循环并发执行,还能共享变量。但是出现新的用法的同时,也带来了新的棘手问题,协程也会泄漏,不恰当的使用会影响性能。下面会逐一介绍各种用法和问题。演示的代码用GO语言写成,因为其简洁明了,而且支持全部功能。

生成器

有的时候,我们需要有一个函数能不断生成数据。比方说这个函数可以读文件,读网络,生成自增长序列,生成随机数。这些行为的特点就是,函数的已知一些变量,如文件路径。然后不断调用,返回新的数据。

下面生成随机数为例, 以让我们做一个会并发执行的随机数生成器。

非并发的做法是这样的:

// 函数 rand_generator_1 ,返回 int

func rand_generator_1() int 
{    
   return rand.Int()
}

上面是一个函数,返回一个int。假如rand.Int()这个函数调用需要很长时间等待,那该函数的调用者也会因此而挂起。所以我们可以创建一个协程,专门执行rand.Int()。

// 函数 rand_generator_2,返回 通道(Channel)

func rand_generator_2() chan int {

// 创建通道

out := make(chan int)

// 创建协程

go func() {

for {

//向通道内写入数据,如果无人读取会等待

out <- rand.Int()

}

}()

return out

}

func main() {

// 生成随机数作为一个服务

rand_service_handler := rand_generator_2()

// 从服务中读取随机数并打印

fmt.Printf("%d\n", <-rand_service_handler)

}

上面的这段函数就可以并发执行了rand.Int()。有一点值得注意到函数的返回可以理解为一个“服务”。但我们需要获取随机数据 时候,可以随时向这个服务取用,他已经为我们准备好了相应的数据,无需等待,随要随到。如果我们调用这个服务不是很频繁,一个协程足够满足我们的需求了。但如果我们需要大量访问,怎么办?我们可以用下面介绍的多路复用技术,启动若干生成器,再将其整合成一个大的服务。

调用生成器,可以返回一个“服务”。可以用在持续获取数据的场合。用途很广泛,读取数据,生成ID,甚至定时器。这是一种非常简洁的思路,将程序并发化。

多路复用

多路复用是让一次处理多个队列的技术。Apache使用处理每个连接都需要一个进程,所以其并发性能不是很好。而Nighx使用多路复用的技术,让一个进程处理多个连接,所以并发性能比较好。同样,在协程的场合,多路复用也是需要的,但又有所不同。多路复用可以将若干个相似的小服务整合成一个大服务。

那么让我们用多路复用技术做一个更高并发的随机数生成器吧。

// 函数 rand_generator_3 ,返回通道(Channel)

func rand_generator_3() chan int {

// 创建两个随机数生成器服务

rand_generator_1 := rand_generator_2()

rand_generator_2 := rand_generator_2()

//创建通道

out := make(chan int)

//创建协程

go func() {

for {

//读取生成器1中的数据,整合

out <- <-rand_generator_1

}

}()

go func() {

for {

//读取生成器2中的数据,整合

out <- <-rand_generator_2

}

}()

return out

}

上面是使用了多路复用技术的高并发版的随机数生成器。通过整合两个随机数生成器,这个版本的能力是刚才的两倍。虽然协程可以大量创建,但是众多协程还是会争抢输出的通道。Go语言提供了Select关键字来解决,各家也有各家窍门。加大输出通道的缓冲大小是个通用的解决方法。

多路复用技术可以用来整合多个通道。提升性能和操作的便捷。配合其他的模式使用有很大的威力。

Furture技术

Furture是一个很有用的技术,我们常常使用Furture来操作线程。我们可以在使用线程的时候,可以创建一个线程,返回Furture,之后可以通过它等待结果。 但是在协程环境下的Furtue可以更加彻底,输入参数同样可以是Furture的。

调用一个函数的时候,往往是参数已经准备好了。调用协程的时候也同样如此。但是如果我们将传入的参数设为通道,这样我们就可以在不准备好参数的情况下调用函数。这样的设计可以提供很大的自由度和并发度。函数调用和函数参数准备这两个过程可以完全解耦。下面举一个用该技术访问数据库的例子。

//一个查询结构体

type query struct {

//参数Channel

sql chan string

//结果Channel

result chan string

}

//执行Query

func execQuery(q query) {

//启动协程

go func() {

//获取输入

sql := <-q.sql

//访问数据库,输出结果通道

q.result <- "get " + sql

}()

}

func main() {

//初始化Query

q :=

query{make(chan string, 1), make(chan string, 1)}

//执行Query,注意执行的时候无需准备参数

execQuery(q)

//准备参数

q.sql <- "select * from table"

//获取结果

fmt.Println(<-q.result)

}

上面利用Furture技术,不单让结果在Furture获得,参数也是在Furture获取。准备好参数后,自动执行。Furture和生成器的区别在于,Furture返回一个结果,而生成器可以重复调用。还有一个值得注意的地方,就是将参数Channel和结果Channel定义在一个结构体里面作为参数,而不是返回结果Channel。这样做可以增加聚合度,好处就是可以和多路复用技术结合起来使用。

Furture技术可以和各个其他技术组合起来用。可以通过多路复用技术,监听多个结果Channel,当有结果后,自动返回。也可以和生成器组合使用,生成器不断生产数据,Furture技术逐个处理数据。Furture技术自身还可以首尾相连,形成一个并发的pipe filter。这个pipe filter可以用于读写数据流,操作数据流。

Future是一个非常强大的技术手段。可以在调用的时候不关心数据是否准备好,返回值是否计算好的问题。让程序中的组件在准备好数据的时候自动跑起来。

并发循环

循环往往是性能上的热点。如果性能瓶颈出现在CPU上的话,那么九成可能性热点是在一个循环体内部。所以如果能让循环体并发执行,那么性能就会提高很多。

要并发循环很简单,只有在每个循环体内部启动协程。协程作为循环体可以并发执行。调用启动前设置一个计数器,每一个循环体执行完毕就在计数器上加一个元素,调用完成后通过监听计数器等待循环协程全部完成。

//建立计数器

sem := make(chan int, N);

//FOR循环体

for i,xi := range data {

//建立协程

go func (i int, xi float) {

doSomething(i,xi);

//计数

sem <- 0;

} (i, xi);

}

// 等待循环结束

for i := 0; i < N; ++i { <-sem }

上面是一个并发循环例子。通过计数器来等待循环全部完成。如果结合上面提到的Future技术的话,则不必等待。可以等到真正需要的结果的地方,再去检查数据是否完成。

通过并发循环可以提供性能,利用多核,解决CPU热点。正因为协程可以大量创建,才能在循环体中如此使用,如果是使用线程的话,就需要引入线程池之类的东西,防止创建过多线程,而协程则简单的多。

Chain Filter技术

前面提到了Future技术首尾相连,可以形成一个并发的pipe filter。这种方式可以做很多事情,如果每个Filter都由同一个函数组成,还可以有一种简单的办法把他们连起来。

由于每个Filter协程都可以并发运行,这样的结构非常有利于多核环境。下面是一个例子,用这种模式来产生素数。

// A concurrent prime sieve

package main

// Send the sequence 2, 3, 4, ... to channel 'ch'.

func Generate(ch chan<- int) {

for i := 2; ; i++ {

ch <- i // Send 'i' to channel 'ch'.

}

}

// Copy the values from channel 'in' to channel 'out',

// removing those divisible by 'prime'.

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'.

}

}

}

// The prime sieve: Daisy-chain Filter processes.

func main() {

ch := make(chan int) // Create a new channel.

go Generate(ch) // Launch Generate goroutine.

for i := 0; i < 10; i++ {

prime := <-ch

print(prime, "\n")

ch1 := make(chan int)

go Filter(ch, ch1, prime)

ch = ch1

}

}

上面的程序创建了10个Filter,每个分别过滤一个素数,所以可以输出前10个素数。

Chain-Filter通过简单的代码创建并发的过滤器链。这种办法还有一个好处,就是每个通道只有两个协程会访问,就不会有激烈的竞争,性能会比较好。

共享变量

协程之间的通信只能够通过通道。但是我们习惯于共享变量,而且很多时候使用共享变量能让代码更简洁。比如一个Server有两个状态开和关。其他仅仅希望获取或改变其状态,那又该如何做呢。可以将这个变量至于0通道中,并使用一个协程来维护。

下面的例子描述如何用这个方式,实现一个共享变量。

//共享变量有一个读通道和一个写通道组成

type sharded_var struct {

reader chan int

writer chan int

}

//共享变量维护协程

func sharded_var_whachdog(v sharded_var) {

go func() {

//初始值

var value int = 0

for {

//监听读写通道,完成服务

select {

case value = <-v.writer:

case v.reader <- value:

}

}

}()

}

func main() {

//初始化,并开始维护协程

v := sharded_var{make(chan int), make(chan int)}

sharded_var_whachdog(v)

//读取初始值

fmt.Println(<-v.reader)

//写入一个值

v.writer <- 1

//读取新写入的值

fmt.Println(<-v.reader)

}

这样,就可以在协程和通道的基础上实现一个协程安全的共享变量了。定义一个写通道,需要更新变量的时候,往里写新的值。再定义一个读通道,需要读的时候,从里面读。通过一个单独的协程来维护这两个通道。保证数据的一致性。

一般来说,协程之间不推荐使用共享变量来交互,但是按照这个办法,在一些场合,使用共享变量也是可取的。很多平台上有较为原生的共享变量支持,到底用那种实现比较好,就见仁见智了。另外利用协程和通道,可以还实现各种常见的并发数据结构,如锁等等,就不一一赘述。

协程泄漏

协程和内存一样,是系统的资源。对于内存,有自动垃圾回收。但是对于协程,没有相应的回收机制。会不会若干年后,协程普及了,协程泄漏和内存泄漏一样成为程序员永远的痛呢?一般而言,协程执行结束后就会销毁。协程也会占用内存,如果发生协程泄漏,影响和内存泄漏一样严重。轻则拖慢程序,重则压垮机器。

C和C++都是没有自动内存回收的程序设计语言,但只要有良好的编程习惯,就能解决规避问题。对于协程是一样的,只要有好习惯就可以了。

只有两种情况会导致协程无法结束。一种情况是协程想从一个通道读数据,但无人往这个通道写入数据,或许这个通道已经被遗忘了。还有一种情况是程想往一个通道写数据,可是由于无人监听这个通道,该协程将永远无法向下执行。下面分别讨论如何避免这两种情况。

对于协程想从一个通道读数据,但无人往这个通道写入数据这种情况。解决的办法很简单,加入超时机制。对于有不确定会不会返回的情况,必须加入超时,避免出现永久等待。另外不一定要使用定时器才能终止协程。也可以对外暴露一个退出提醒通道。任何其他协程都可以通过该通道来提醒这个协程终止。

对于协程想往一个通道写数据,但通道阻塞无法写入这种情况。解决的办法也很简单,就是给通道加缓冲。但前提是这个通道只会接收到固定数目的写入。比方说,已知一个通道最多只会接收N次数据,那么就将这个通道的缓冲设置为N。那么该通道将永远不会堵塞,协程自然也不会泄漏。也可以将其缓冲设置为无限,不过这样就要承担内存泄漏的风险了。等协程执行完毕后,这部分通道内存将会失去引用,会被自动垃圾回收掉.

func never_leak(ch chan int) {

//初始化timeout,缓冲为1

timeout := make(chan bool, 1)

//启动timeout协程,由于缓存为1,不可能泄露

go func() {

time.Sleep(1 * time.Second)

timeout <- true

}()

//监听通道,由于设有超时,不可能泄露

select {

case <-ch:

// a read from ch has occurred

case <-timeout:

// the read from ch has timed out

}

}

上面是个避免泄漏例子。使用超时避免读堵塞,使用缓冲避免写堵塞。

和内存里面的对象一样,对于长期存在的协程,我们不用担心泄漏问题。一是长期存在,二是数量较少。要警惕的只有那些被临时创建的协程,这些协程数量大且生命周期短,往往是在循环中创建的,要应用前面提到的办法,避免泄漏发生。协程也是把双刃剑,如果出问题,不但没能提高程序性能,反而会让程序崩溃。但就像内存一样,同样有泄漏的风险,但越用越溜了。

并发模式之实现

在并发编程大行其道的今天,对协程和通道的支持成为各个平台比不可少的一部分。虽然各家有各家的叫法,但都能满足协程的基本要求—并发执行和可大量创建。笔者对他们的实现方式总结了一下。

下面列举一些已经支持协程的常见的语言和平台。

语言/平台

实现时间

协程名称

备注

GoLang

原生支持

goroutines

Erlang

原生支持

process

函数式语言

Scala

原生支持

actor

函数式编程

Python

2.5版本后

coroutine

官方Python不完全实现 Stackless Python支持

Perl

6.0版本后

coroutine

Ruby

1.9 版本后

fiber

Lua

原生支持

coroutine

C#

.net 2.0版本后

fiber

GoLang 和Scala作为最新的语言,一出生就有完善的基于协程并发功能。Erlang最为老资格的并发编程语言,返老还童。其他二线语言则几乎全部在新的版本中加入了协程。

令人惊奇的是C/C++和Java这三个世界上最主流的平台没有在对协程提供语言级别的原生支持。他们都背负着厚重的历史,无法改变,也无需改变。但他们还有其他的办法使用协程。

Java平台有很多方法实现协程:

  • 修改虚拟机:对JVM打补丁来实现协程,这样的实现效果好,但是失去了跨平台的好处
  • 修改字节码:在编译完成后增强字节码,或者使用新的JVM语言。稍稍增加了编译的难度。
  • 使用JNI:在Jar包中使用JNI,这样易于使用,但是不能跨平台。
  • 使用线程模拟协程:使协程重量级,完全依赖JVM的线程实现。

其中修改字节码的方式比较常见。因为这样的实现办法,可以平衡性能和移植性。最具代表性的JVM语言Scala就能很好的支持协程并发。流行的Java Actor模型类库akka也是用修改字节码的方式实现的协程。

对于C语言,协程和线程一样。可以使用各种各样的系统调用来实现。协程作为一个比较高级的概念,实现方式实在太多,就不讨论了。比较主流的实现有libpcl, coro,lthread等等。

对于C++,有Boost实现,还有一些其他开源库。还有一门名为μC++语言,在C++基础上提供了并发扩展。

可见这种编程模型在众多的语言平台中已经得到了广泛的支持,不再小众。如果想使用的话,随时可以加到自己的工具箱中。

结语

本文探讨了一个极其简洁的并发模型。在只有协程和通道这两个基本元件的情况下。可以提供丰富的功能,解决形形色色实际问题。而且这个模型已经被广泛的实现,成为潮流。相信这种并发模型的功能远远不及此,一定也会有更多更简洁的用法出现。或许未来CPU核心数目将和人脑神经元数目一样多,到那个时候,我们又要重新思考并发模型了。

原文发布于微信公众号 - Golang语言社区(Golangweb)

原文发表时间:2017-05-12

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏小灰灰

Quick-Task 动态脚本支持框架之结构设计篇

文章链接:https://liuyueyi.github.io/hexblog/2018/07/23/180723-Quick-Task-动态脚本支持框架之结构...

1043
来自专栏FreeBuf

WireShark+Winhex:流量分析的好搭档

这篇文章你将学会的知识点有 1、进阶的wireshark的流量分析、解码、追踪流、导出文件 2、利用hackbar进行base64、URL编码转换 3、利用wi...

7346
来自专栏码洞

Github上最受欢迎的Python框架Flask入门

flask最近终于发布了它的1.0版本更新,从项目开源到最近的1.0版本flask已经走过了8个年头。

8402
来自专栏小灰灰

Quick-Task 动态脚本支持框架之任务动态加载

前面几篇博文分别介绍了整个项目的基本架构,使用说明,以及整体框架的设计与实现初稿,接下来则进入更细节的实现篇,将整个工程中核心实现捞出来,从为什么这么设计到最终...

1292
来自专栏Java架构沉思录

不懂RPC,休谈微服务

在学校期间大家都写过不少程序,比如写个hello world服务类,然后本地调用下,如下所示。这些程序的特点是服务消费方和服务提供方是本地调用关系。

1332
来自专栏Web 开发

最近弄HybridApp的一些心得

SDK的项目折腾了一个月,终于快到收获的时候,把这过程中的一些心得体会记录一下吧~

900
来自专栏owent

关于BUS通信系统的一些思考(二)

虽然我很不愿意再设计一套BUS系统,但是现有的一些确实都没有特别符合我的口味的。所以还是尝试设计一个出来。

973
来自专栏IT派

用Click编写Python命令行工具

在编写Python命令行(CLI)应用程序时,使用Click库进行参数解析的深入教程

2051
来自专栏信安之路

用150行python代码来做代码审计笔记

通过审计源代码,也就是查看源代码,来发现其中存在的隐患,代码审计需要对被审计的语言有充分的了解,不仅是能读懂源代码,还要了解语言本身的缺陷。很多时候代码审计的突...

1010
来自专栏微信公众号:Java团长

Java后端程序员1年工作经验总结

  毕业已经一年有余,这一年里特别感谢技术管理人员的器重,以及同事的帮忙,学到了不少东西。这一年里走过一些弯路,也碰到一些难题,也受到过做为一名开发却经常为系统...

2032

扫码关注云+社区

领取腾讯云代金券