Go语言之Go语言并发

Go 语言并发

Golang从语言层面就对并发提供了支持,而goruntine是Go语言并发设计的核心。

Go语言的并发机制运用起来非常舒适,在启动并发的方式上直接添加了语言级的关键字就可以实现,和其他编程语言相比更加轻量。

进程&线程

A、进程是程序在操作系统中的一次执行过程,系统进行资源分配和调度的一个独立单位。

B、线程是进程的一个执行实体,是CPU调度和分派的基本单位,它是比进程更小的能独立运行的基本单位。

C、一个进程可以创建和撤销多个线程;同一个进程中的多个线程之间可以并发执行。

并发&并行

A、多线程程序在一个核的cpu上运行,就是并发。

B、多线程程序在多个核的cpu上运行,就是并行。

并发不是并行:

并发主要由切换时间片来实现"同时"运行,并行则是直接利用多核实现多线程的运行,Go程序可以设置使用核数,以发挥多核计算机的能力。

协程&线程

协程:独立的栈空间,共享堆空间,调度由用户自己控制,本质上有点类似于用户级线程,这些用户级线程的调度也是自己实现的。

线程:一个线程上可以跑多个协程,协程是轻量级的线程。

Goroutine 介绍

goroutine 只是由官方实现的超级"线程池"。每个实力4~5KB的栈内存占用和由于实现机制而大幅减少的创建和销毁开销是Go语言高并发的根本原因。

goroutine 奉行通过通信来共享内存,而不是共享内存来通信。只需在函数调用语句前添加 go 关键字,就可创建并发执行单元。开发人员无需了解任何执行细节,调度器会自动将其安排到合适的系统线程上执行。goroutine 是一种非常轻量级的实现,可在单个进程里执行成千上万的并发任务。

事实上,入口函数 main 就以 goroutine 运行。另有与之配套的 channel 类型,用以实现 "以通讯来共享内存" 的 CSP 模式。

goroutine 是通过 Go 的 runtime管理的一个线程管理器

package main

import (
	"fmt"
	"time"
)

func main() {
	go func() {
		fmt.Println("hello word")
	}()
	time.Sleep(1 * time.Second)
}
进入 main 函数开启一个 goroutine 运行匿名函数函数体内容:fmt.Println("Hello, World!") 。主线程执行 time.Sleep(1 * time.Second) 等待 1 秒。goroutine 执行完毕回到主线程,主线程的sleep 完成结束程序。

注意:若去掉 time.Sleep(1 * time.Second) 这段代码,进入 main 函数开启一个 goroutine,没等 goroutine 运行匿名函数函数体内容,主线程已经完成结束程序。

Go语言Chan应用

Channel 是 CSP 模式的具体实现,用于多个 goroutine 通讯。其内部实现了同步,确保并发安全。
Channel 是先进先出,线程安全的,多个goroutine同时访问,不需要加锁。

chan 阻塞

我们定义的管道 intChan 容量是5,开启 goroutine 写入10条数据,在写满5条数据时会阻塞,而 read() 每秒会从 intChan 中读取一条,然后write() 再会写入一条数据。

package main

import (
	"fmt"
	"time"
)

func write(ch chan int) {
	for i := 0; i < 10; i++ {
		ch <- i
		fmt.Println("write data:", i)
	}
}
func read(ch chan int) {
	for {
		i := <-ch
		fmt.Println("read data:", i)
		time.Sleep(time.Second)
	}
}
func main() {
	intChan := make(chan int, 5)
	go write(intChan)
	go read(intChan)

	time.Sleep(10 * time.Second)
}

同步模式

默认为同步模式,需要发送和接收配对。否则会被阻塞,直到另一方准备好后被唤醒。

package main

import "fmt"

func main() {
	data := make(chan string) // 数据交换队列
	exit := make(chan bool)   // 退出通知

	go func() {
		for d := range data { // 从队列迭代接收数据,直到 close 。
			fmt.Println(d)
		}
		fmt.Println("received over")
		exit <- true // 发出退出通知。
	}()
	data <- "oldboy" // 发送数据。
	data <- "Linux"
	data <- "GOlang"
	data <- "python"
	close(data) // 关闭队列。
	fmt.Println("send over")
	<-exit // 等待退出通知。
}

异步模式

异步方式通过判断缓冲区来决定是否阻塞。如果缓冲区已满,发送被阻塞;缓冲区为空,接收被阻塞。

通常情况下,异步 channel 可减少排队阻塞,具备更高的效率。但应该考虑使用指针规避大对象拷贝,将多个元素打包,减小缓冲区大小。

package main

import "fmt"

func main() {
	data := make(chan string, 3) // 缓冲区可以存储 3 个元素
	exit := make(chan bool)

	data <- "old boy" // 在缓冲区未满前,不会阻塞。
	data <- "python"
	data <- "linux"

	go func() {
		for d := range data { // 在缓冲区未空前,不会阻塞。
			fmt.Println(d)
		}
		//  表示读取出data通道中数据
		exit <- true
	}()
	data <- "java" // 如果缓冲区已满,阻塞。
	data <- "C"
	close(data)
	<-exit
}

chan 选择

如果需要同时处理多个 channel,可使用 select 语句。它随机选择一个可用 channel 做收发操作,或执行 default case。

用 select 实现超时控制:

package main

import (
	"fmt"
	"time"
)

func main() {
	exit := make(chan bool)
	intChan := make(chan int, 2)
	strChan := make(chan string, 2)

	go func() {
		select {
		case vi := <-intChan:
			fmt.Println(vi)
		case vs := <-strChan:
			fmt.Println(vs)
		case <-time.After(time.Second * 3):
			fmt.Println("timeout.")
		}

		exit <- true
	}()

	// intChan <- 100 // 注释掉,引发 timeout。
	// strChan <- "oldboy"

	<-exit
}

在循环中使用 select default case 需要小心,避免形成洪水。

简单工厂模式

用简单工厂模式打包并发任务和 channel。

package main

import (
	"fmt"
	"math/rand"
	"time"
)

func NewTest() chan int {
	c := make(chan int)
	rand.Seed(time.Now().UnixNano())
	go func() {
		time.Sleep(time.Second)
		c <- rand.Int()
	}()
	return c
}
func main() {
	t := NewTest()
	fmt.Println(<-t) // 等待 goroutine 结束返回。
}

Go 语言WaitGroup

WaitGroup能够一直等到所有的goroutine执行完成,并且阻塞主线程的执行,直到所有的goroutine执行完成。

WaitGroup总共有三个方法:Add(delta int),Done(),Wait()。简单的说一下这三个方法的作用。

Add:添加或者减少等待goroutine的数量;

Done:相当于Add(-1);

Wait:执行阻塞,直到所有的WaitGroup数量变成 0;

WaitGroup用于线程同步,WaitGroup等待一组线程集合完成,才会继续向下执行。 主线程(goroutine)调用Add来设置等待的线程(goroutine)数量。 然后每个线程(goroutine)运行,并在完成后调用Done。 同时,Wait用来阻塞,直到所有线程(goroutine)完成才会向下执行。

WaitGroup实例如下:

package main

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

func main() {
    var wg sync.WaitGroup
    for i := 0; i < 5; i++ {
        wg.Add(1)
        go func(n int) {
            // defer wg.Done()
            defer wg.Add(-1)
            EchoNum(n)
        }(i)
    }
    wg.Wait()
}

func EchoNum(i int) {
    time.Sleep(time.Second)
    fmt.Println(i)
}

程序中将每次循环的数量 sleep 1 秒钟后输出。如果程序不使用WaitGroup,将不会输出结果。因为goroutine还没执行完,主线程已经执行完毕。
注掉的 defer wg.Done() 和 defer wg.Add(-1) 作用一样。

WaitGroup应用

一、用 channel 实现信号量 (semaphore)。

package main

import (
	"sync"
)

func main() {
	wg := sync.WaitGroup{}
	wg.Add(3) //增加三个线程
	sem := make(chan int, 1)
	for i := 0; i < 3; i++ {
		go func(id int) {
			defer wg.Done() //减少一个线程
			sem <- 1        // 向 sem 发送数据,阻塞或者成功。
			for x := 0; x < 3; x++ {
				println(id, x)
			}
			<-sem // 接收数据,使得其他阻塞 goroutine 可以发送数据
		}(i)
	}
	wg.Wait()

}
D:\goprogram\go\src\day10       
λ go run test.go

二、用 closed channel 发出退出通知。

package main

import (
	"sync"
	"time"
)

func main() {
	wg := sync.WaitGroup{}
	exit := make(chan bool)
	for i := 0; i < 2; i++ {
		wg.Add(1)
		go func(n int) {
			defer wg.Done()
			task := func() {
				println(n, time.Now().String())
				time.Sleep(1 * time.Second)
			}
			for {
				select {
				case <-exit: // closed channel 不会阻塞,因此可用作退出通知。
					return
				default: // 执行正常任务。
					task()
				}
			}
		}(i)
	}
	time.Sleep(time.Second * 3) // 让测试 goroutine 运行一会。
	close(exit)                 // 发出退出通知。
	wg.Wait()
}

WaitGroup陷阱

一、add 数量小于done数量导致 WaitGroup数量为负数

package main

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

func main() {
    var wg sync.WaitGroup
    wg.Add(1)

    oldboy := func() {
        time.Sleep(time.Second)
        fmt.Println("The old boy welcomes you.")
        wg.Done()
    }

    go oldboy()
    go oldboy()
    go oldboy()

    time.Sleep(time.Second * 3)
    wg.Wait()
}

运行错误:

panic: sync: negative WaitGroup counter

二、add 数量大于 done 数量造成 deadlock

package main

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

func main() {
    var wg sync.WaitGroup
    wg.Add(4)

    oldboy := func() {
        time.Sleep(time.Second)
        fmt.Println("The old boy welcomes you.")
        wg.Done()
    }

    go oldboy()
    go oldboy()
    go oldboy()

    time.Sleep(time.Second * 3)
    wg.Wait()
}

运行错误:

fatal error: all goroutines are asleep - deadlock!

三、跳过 add 和 Done 操作,直接执行 Wait

package main

import (
    "fmt"
    "sync"
)

func main() {
    wg := sync.WaitGroup{}

    for i := 0; i < 5; i++ {
        go func(wg sync.WaitGroup, i int) {
            wg.Add(1)
            fmt.Printf("i=>%d\n", i)
            wg.Done()
        }(wg, i)
    }
    wg.Wait()
    fmt.Println("exit")
}

WaitGroup 同步的是 goroutine, 而上面的代码却在 goroutine 中进行 Add(1) 操作。因此,可能在这些 goroutine 还没来得及 Add(1) 就已经执行 Wait 操作了。

四、WaitGroup 拷贝传值问题

package main

import (
    "fmt"

    "sync"
)

func main() {
    wg := sync.WaitGroup{}

    for i := 0; i < 5; i++ {
        wg.Add(1)
        go func(wg sync.WaitGroup, i int) {
            fmt.Printf("i=>%d\n", i)
            wg.Done()
        }(wg, i)
    }
    wg.Wait()
}

运行错误:

fatal error: all goroutines are asleep - deadlock!

wg 给拷贝传递到了 goroutine 中,导致只有 Add 操作,其实 Done操作是在 wg 的副本执行的,因此 Wait 就死锁了。

正确代码实例如下:

package main

import (
    "fmt"

    "sync"
)

func main() {
    wg := new(sync.WaitGroup)
    // wg := &sync.WaitGroup{}

    for i := 0; i < 5; i++ {
        wg.Add(1)
        go func(wg *sync.WaitGroup, i int) {
            fmt.Printf("i=>%d\n", i)
            wg.Done()
        }(wg, i)
    }
    wg.Wait()
}

Go 语言runtime

runtime包提供Go语言运行时的系统交互的操作,例如控制goruntine的功能。

调度器不能保证多个 goroutine 执行次序,且进程退出时不会等待它们结束。

默认情况下,进程启动后仅允许一个系统线程服务于 goroutine。可使用环境变量或标准库函数 runtime.GOMAXPROCS 修改,让调度器用多个线程实现多核并行,而不仅仅是并发。

runtime包常用方法

const GOOS string = theGoos

GOOS是可执行程序的目标操作系统(将要在该操作系统的机器上执行):darwin、freebsd、linux等。

func Gosched()

Gosched使当前go程放弃处理器,以让其它go程运行。它不会挂起当前go程,因此当前go程未来会恢复执行。

func NumCPU() int

NumCPU返回本地机器的逻辑CPU个数。

func GOROOT() string

GOROOT返回Go的根目录。如果存在GOROOT环境变量,返回该变量的值;否则,返回创建Go时的根目录。

func GOMAXPROCS(n int) int

GOMAXPROCS设置可同时执行的最大CPU数,并返回先前的设置。 若 n < 1,它就不会更改当前设置。本地机器的逻辑CPU数可通过 NumCPU 查询。本函数在调度程序优化后会去掉。

func Goexit()

Goexit终止调用它的go程。其它go程不会受影响。Goexit会在终止该go程前执行所有defer的函数。

在程序的main go程调用本函数,会终结该go程,而不会让main返回。因为main函数没有返回,程序会继续执行其它的go程。如果所有其它go程都退出了,程序就会崩溃。

func NumGoroutine() int

NumGoroutine返回当前存在的Go程数。

runtime包应用

一、查看机器的逻辑CPU个数、Go的根目录、操作系统

package main

import "runtime"

func main() {
	println("cpu:", runtime.NumCPU())
	println("go:", runtime.GOROOT())
	println("操作系统:", runtime.GOOS)

}
D:\goprogram\go\src\day10
λ go run test.go
cpu: 4
go: D:\go\go
操作系统: windows

二、GOMAXPROCS 设置golang运行的cpu核数

Golang 默认所有任务都运行在一个 cpu 核里,如果要在 goroutine 中使用多核,可以使用 runtime.GOMAXPROCS 函数修改,当参数小于 1 时使用默认值。

package main

import (
    "fmt"
    "runtime"
)

var (
    signal = false
)

func oldboy() {
    signal = true
}

func init() {
    runtime.GOMAXPROCS(1)
}

func main() {
    go oldboy()
    for {
        if signal {
            break
        }
    }
    fmt.Println("end")
}

上述代码单核执行如果for前面或者中间不延迟,主线程不会让出CPU,导致异步的线程无法执行,从而无法设置signal的值,从而出现死循环。

运行的cpu核数设置成2核

package main

import (
    "fmt"
    "runtime"
)

var (
    signal = false
)

func oldboy() {
    signal = true
}

func init() {
    runtime.GOMAXPROCS(2)
}

func main() {
    go oldboy()
    for {
        if signal {
            break
        }
    }
    fmt.Println("end")
}

运行结果:

end

三、Gosched 让当前的 goroutine 让出 CPU

这个函数的作用是让当前 goroutine 让出 CPU,当一个 goroutine 发生阻塞,Go 会自动地把与该 goroutine 处于同一系统线程的其他 goroutine 转移到另一个系统线程上去,以使这些 goroutine 不阻塞。当前的 goroutine 不会挂起,当前的 goroutine 程未来会恢复执行。

runtime.Gosched()用于让出CPU时间片。这就像跑接力赛,A跑了一会碰到代码runtime.Gosched()就把接力棒交给B了,A歇着了,B继续跑。

package main

import (
	"runtime"
	"sync"
)

func main() {
	wg := new(sync.WaitGroup)
	wg.Add(1)
	go func() {
		for i := 0; i < 6; i++ {
			println(i)
			runtime.Gosched()
		}
		defer wg.Done()
	}()
	for i := 0; i < 6; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			println("Hello.Golang!")
		}()
	}
	wg.Wait()
}
D:\goprogram\go\src\day10
λ go run test.go
0
1
2
3
4
5
Hello.Golang!
Hello.Golang!
Hello.Golang!
Hello.Golang!
Hello.Golang!
Hello.Golang!

四、Goexit 终止当前 goroutine 执行

调用 runtime.Goexit 将立即终止当前 goroutine 执行,调度器确保所有已注册 defer 延迟调用被执行。

package main

import (
    "fmt"
    "runtime"
    "sync"
)

func main() {
    wg := new(sync.WaitGroup)
    wg.Add(1)

    go func() {
        defer wg.Done()
        defer fmt.Println("A.defer")
        func() {
            defer fmt.Println("B.defer")
            runtime.Goexit() // 终止当前 goroutine
            fmt.Println("B") // 不会执行
        }()

        fmt.Println("A") // 不会执行
    }()

    wg.Wait()
}
B.defer
A.defer

相关推荐