Go程序设计3——并发编程

时间：2022-04-28 04:22:35

1 channel

　　一般channel的声明形式为：

var chanName chan ElementType
　　与一般的变量声明不同的地方仅仅是在类型之前增加了chan关键字。ElementType指定这个channel所能传递的元素类型。举个例子，我们声明一个传递类型的int的channel:
var ch chan int
　　或者声明一个map，元素是bool型的channel
var m map[String] chan bool
　　定义一个channel也很简单，直接使用内置的函数make()即可:
ch := make(chan int)
　　这就声明并初始化了一个int型的名为ch的channel。在channel的用法中，最常见的包括写入和读出，将一个数据写入至channel的语法很直观。
ch <- value
　　向channel写入数据通常会导致程序阻塞，直到有其他goroutine从这个channel中读取数据。从channel中读取数据的语法是:
value := <- ch
　　如果channel之前没有写入数据，那么从channel中读取数据也会导致程序阻塞，直到channel中被写入数据为止。我们之后还会提到如何控制channel只接受写或者只允许读取，即单向channel。

1.1 select

　　早在Unix时代，select机制就已经被引入，通过调用select函数来监控一系列的文件句柄，一旦其中一个文件句柄发生了IO动作，该select调用就会被返回。后来该机制也被用于实现高并发的Socket服务器程序。Go语言直接在语言级别支持select关键字，用于处理异步IO问题。
　　select的用法与switch语言非常类似，由select开始一个新的选择块，每个选择条件由case语句来描述，与switch语句可以选择任何可使用相等比较的条件相比，select有比较多的限制，其中最大的一条限制就是每个case语句里必须是一个IO操作，大致结构如下：

select {

 case <-chan1:

 // 如果chan1成功读到数据，则进行该case处理语句

 case chan2 <- 1:

 // 如果成功向chan2写入数据，则进行该case处理语句

  default:

 // 如果上面都没有成功，则进入default处理流程

}

可以看出，select不像switch，后面并不带判断条件，而是直接去查看case语句。每个case语句都必须是一个面向channel的操作。比如上面的例子中，第一个case试图从chan1读取
一个数据并直接忽略读到的数据，而第二个case则是试图向chan2中写入一个整型数1，如果这两者都没有成功，则到达default语句。

ch := make(chan int, 1)

for{

    select{

        case ch <- 0:

        case ch <- 1:

    }

    i := <- ch

    fmt.Println("Value received:", i)

}

1.2 缓冲机制

　　之前我们示范创建的都是不带缓冲的channel，这种做法对于传递单个数据的场景可以接受，但对于需要持续传输大量数据的场景就有些不合适了。接下来我们介绍如何给channel带上缓冲，从而达到消息队列的效果。 要创建一个带缓冲的channel，其实也非常容易：
c := make(chan int, 1024)
　　在调用make()时将缓冲区大小作为第二个参数传入即可，比如上面这个例子就创建了一个大小为1024的int类型channel，即使没有读取方，写入方也可以一直往channel里写入，在缓冲区被填完之前都不会阻塞从带缓冲的channel中读取数据可以使用与常规非缓冲channel完全一致的方法，但我们也可以使用range关键来实现更为简便的循环读取：

for i := range c {

    fmt.Println("Received:", i)

}

1.3 channel的传递

　　需要注意的是，在Go语言中channel本身也是一个原生类型，与map之类的类型地位一样，因此channel本身在定义后也可以通过channel来传递。channel和Unix的管道pipe很类似。下面我们利用channel可被传递的特性来实现我们的管道，为了简化表达，假设在管道中传递的数据只是一个整型数，在实际场景中通常是一个数据块。
首先限定基本的数据结构：

type PipeData struct{

    value int

    handler func(int) int

    next chan int

}

1.4 单向channel

　　单向channel只能用于发送或者接受数据，channel本身必然是同时支持读写的，否则根本没法用。如果只允许读，那就没法写数据，所以channel是空的，如果只允许写，没法读，那没有意义，因为没法读。
　　单向channel变量的声明非常简单，如下：

var ch1 chan int // ch1是一个正常的channel，不是单向的
var ch2 chan<- float64// ch2是单向channel，只用于写float64数据
var ch3 <-chan int // ch3是单向channel，只用于读取int数据
关闭channel
　　关闭channel很简单，使用close函数
close(ch)

2 多核并行化

　　在执行一些昂贵的计算任务时，我们希望能够尽量利用现代服务器普遍具备的多核特性来尽量将任务并行化，从而达到降低总计算时间的目的。此时我们需要了解CPU核心的数量，并针对
性地分解计算任务到多个goroutine中去并行运行。

　　下面来模拟一个完全可以并行的计算任务，计算N个整型数的总和，我们可以将所有整型数分成M份，M即CPU的个数，让每个CPU开始计算分给它的那份计算任务，最后将每个CPU的计算结果做累加，得到所有N个整型数的总和：

type Vector[] float64

//分配给每个CPU的计算任务

func (v Vector) DoSome(i, n int, u Vector, c chan int){

    for ; i < n; i++{

        v[i] += u.Op(v[i])

    }

    c <-

}

const NCPU =     // 假设总共有16核 

func (v Vector) DoAll(u Vector) { 

    c := make(chan int, NCPU)  // 用于接收每个CPU的任务完成信号 

 for i := ; i < NCPU; i++ {

  go v.DoSome(i*len(v)/NCPU, (i+)*len(v)/NCPU, u, c)

    } 

 // 等待所有CPU的任务完成

 for i := ; i < NCPU; i++ {

  <-c    // 获取到一个数据，表示一个CPU计算完成了

    }

 // 到这里表示所有计算已经结束

}

　　这两个函数看起来设计非常合理。DoAll()会根据CPU核心的数目对任务进行分割，然后开辟多个goroutine来并行执行这些计算任务。执行后，计算时间并没有降到原来的1/N，答案是否定的，因为当前版本还没有支持。还是单核的。官方的答案是，这是当前版本的Go编译器还不能很智能地去发现和利用多核的优势。虽然我们确实创建了多个goroutine，并且从运行状态看这些goroutine也都在并行运行，但实际上所有这些goroutine都运行在同一个CPU核心上，在一个goroutine得到时间片执行的时候，其他goroutine都会处于等待状态。从这一点可以看出，虽然goroutine简化了我们写并行代码的过程，但实际上整体运行效率并不真正高于单线程程序。
在Go语言升级到默认支持多CPU的某个版本之前，我们可以先通过设置环境变量
　　GOMAXPROCS的值来控制使用多少个CPU核心。具体操作方法是通过直接设置环境变量GOMAXPROCS的值，或者在代码中启动goroutine之前先调用以下这个语句以设置使用16个CPU核心：
runtime.GOMAXPROCS(16)
　　到底应该设置多少个CPU核心呢，其实runtime包中还提供了另外一个函数NumCPU()来获取核心数。可以看到，Go语言其实已经感知到所有的环境信息，下一版本中完全可以利用这些
信息将goroutine调度到所有CPU核心上，从而最大化地利用服务器的多核计算能力。抛弃GOMAXPROCS只是个时间问题。

相关文章