1 Go语言并行编程基础

1.1 Go并发模型简介

Go 语言的并发模型是基于 CSP (Communicating Sequential Processes) 的，由 Tony Hoare 在 1978 年提出。在 Go 中，这个模型通过 goroutine 和 channel 的使用得以实现。goroutine 是轻量级的线程，由 Go 运行时调度，而 channel 则是 goroutine 之间通信的管道，保证了数据的安全交换。

1.1.1 goroutine 的创建与调度

goroutine 的创建非常简单，只需在函数调用前加上 go 关键字即可。Go 运行时会负责调度这些 goroutine，使得它们能够并行执行。

package main
import (
"fmt"
"time"
)
// 定义一个简单的函数，用于在 goroutine 中执行
func say(s string) {
for i := 0; i < 5; i++ {
time.Sleep(100 * time.Millisecond)
fmt.Println(s)
}
}
func main() {
// 创建两个 goroutine
go say("world")
go say("hello")
// 主 goroutine 会等待其他 goroutine 完成，但这里我们使用 time.Sleep 来确保其他 goroutine 有时间运行
time.Sleep(1 * time.Second)
}

1.1.2 channel 的使用

channel 是 goroutine 之间通信的管道，可以发送和接收数据。channel 的使用保证了并发程序中的数据安全，避免了共享内存的竞态条件。

package main
import (
"fmt"
)
// 定义一个整型 channel
ch := make(chan int)
func writer() {
// 向 channel 中写入数据
for i := 0; i < 10; i++ {
ch <- i
}
close(ch)
}
func reader() {
// 从 channel 中读取数据
for {
num, ok := <-ch
if !ok {
break
}
fmt.Println(num)
}
}
func main() {
go writer()
go reader()
// 主 goroutine 等待其他 goroutine 完成
time.Sleep(1 * time.Second)
}

1.2 goroutine和channel的使用

goroutine 和 channel 的结合使用是 Go 并发编程的核心。goroutine 负责执行任务，而 channel 则负责在 goroutine 之间传递数据和同步。

1.2.1 使用 channel 进行数据传递

package main
import (
"fmt"
)
func producer(ch chan int) {
for i := 0; i < 10; i++ {
ch <- i
}
close(ch)
}
func consumer(ch chan int) {
for num := range ch {
fmt.Println(num)
}
}
func main() {
ch := make(chan int)
go producer(ch)
go consumer(ch)
}

1.2.2 使用 channel 进行同步

channel 也可以用于同步 goroutine 的执行。例如，一个 goroutine 可以在完成任务后通过 channel 发送一个信号，另一个 goroutine 接收到信号后继续执行。

package main
import (
"fmt"
)
func taskDone(ch chan bool) {
// 模拟一个耗时的任务
time.Sleep(1 * time.Second)
ch <- true
}
func main() {
ch := make(chan bool)
go taskDone(ch)
<-ch // 等待 taskDone 完成
fmt.Println("任务完成")
}

1.3 并发控制：sync包详解

Go 的 sync 包提供了多种工具来控制并发，包括互斥锁、读写锁、等待组和原子操作。

1.3.1 互斥锁 Mutex

互斥锁用于保护共享资源，确保同一时间只有一个 goroutine 能够访问。

package main
import (
"fmt"
"sync"
"time"
)
var count int = 0
var mutex sync.Mutex
func increment() {
mutex.Lock()
count++
mutex.Unlock()
}
func main() {
var wg sync.WaitGroup
for i := 0; i < 1000; i++ {
wg.Add(1)
go func() {
defer wg.Done()
increment()
}()
}
wg.Wait()
fmt.Println(count)
}

1.3.2 读写锁 RwLocker

读写锁允许多个读操作同时进行，但写操作是独占的。

package main
import (
"fmt"
"sync"
"time"
)
type Data struct {
sync.RWMutex
value int
}
func (d *Data) read() {
d.RLock()
defer d.RUnlock()
fmt.Println(d.value)
}
func (d *Data) write(newValue int) {
d.Lock()
defer d.Unlock()
d.value = newValue
}
func main() {
data := &Data{value: 10}
for i := 0; i < 10; i++ {
go data.read()
}
go data.write(20)
time.Sleep(1 * time.Second)
}

1.3.3 等待组 WaitGroup

等待组用于等待一组 goroutine 完成。

package main
import (
"fmt"
"sync"
"time"
)
func task(wg *sync.WaitGroup) {
defer wg.Done()
time.Sleep(1 * time.Second)
fmt.Println("任务完成")
}
func main() {
var wg sync.WaitGroup
for i := 0; i < 10; i++ {
wg.Add(1)
go task(&wg)
}
wg.Wait()
fmt.Println("所有任务完成")
}

1.3.4 原子操作 Atomic

原子操作用于在不使用锁的情况下更新共享变量。

package main
import (
"fmt"
"sync/atomic"
"time"
)
var count int64 = 0
func increment() {
atomic.AddInt64(&count, 1)
}
func main() {
for i := 0; i < 1000; i++ {
go increment()
}
time.Sleep(1 * time.Second)
fmt.Println(count)
}

通过以上示例，我们可以看到 Go 语言如何通过 goroutine 和 channel 实现并行编程，以及 sync 包如何帮助我们控制并发，确保程序的正确性和效率。

2 Go语言并行编程进阶

2.1 并发模式：生产者-消费者模式

2.1.1 原理

生产者-消费者模式是并发编程中一种常见的模式，它通过将任务的产生（生产者）和处理（消费者）分离，来实现资源的有效利用和系统的高吞吐量。在Go语言中，这一模式可以通过goroutine和channel来实现。生产者将数据发送到channel，而消费者从channel接收数据并处理。

2.1.2 内容

2.1.2.1 示例代码

package main
import (
"fmt"
"sync"
)
// 定义一个channel用于生产者和消费者之间的通信
var ch = make(chan int, 10)
// 生产者函数，生成数据并发送到channel
func producer(wg *sync.WaitGroup) {
defer wg.Done()
for i := 0; i < 10; i++ {
ch <- i
fmt.Printf("生产者生成数据: %d\n", i)
}
close(ch)
}
// 消费者函数，从channel接收数据并处理
func consumer(wg *sync.WaitGroup) {
defer wg.Done()
for {
select {
case data, ok := <-ch:
if !ok {
fmt.Println("消费者接收数据完成")
return
}
fmt.Printf("消费者处理数据: %d\n", data)
}
}
}
func main() {
var wg sync.WaitGroup
// 启动生产者
wg.Add(1)
go producer(&wg)
// 启动消费者
wg.Add(1)
go consumer(&wg)
// 等待所有goroutine完成
wg.Wait()
}

2.1.2.2 代码解释

• ch是一个缓冲大小为10的channel，用于生产者和消费者之间的数据传递。
• producer函数是一个生产者，它生成0到9的整数，并通过ch <- i将数据发送到channel。
• consumer函数是一个消费者，它从channel接收数据。当channel关闭时，consumer函数结束。
• sync.WaitGroup用于同步goroutine，确保所有goroutine完成后再结束main函数。

2.2 并发模式：工作窃取模式

2.2.1 原理

工作窃取模式是一种任务调度策略，其中多个工作队列被维护，每个goroutine都有自己的队列。当一个goroutine完成自己的工作时，它可以从其他队列“窃取”工作来执行，从而避免了资源的闲置和提高了系统的整体效率。

2.2.2 内容

2.2.2.1 示例代码

package main
import (
"fmt"
"sync"
)
// 定义一个工作队列
type WorkQueue struct {
sync.Mutex
tasks []int
}
// 向工作队列中添加任务
func (wq *WorkQueue) AddTask(task int) {
wq.Lock()
defer wq.Unlock()
wq.tasks = append(wq.tasks, task)
}
// 从工作队列中获取任务
func (wq *WorkQueue) GetTask() (int, bool) {
wq.Lock()
defer wq.Unlock()
if len(wq.tasks) == 0 {
return 0, false
}
task := wq.tasks[0]
wq.tasks = wq.tasks[1:]
return task, true
}
// 工作窃取函数，从自己的队列或邻近队列窃取任务
func worker(wq *WorkQueue, wg *sync.WaitGroup) {
defer wg.Done()
for {
task, ok := wq.GetTask()
if !ok {
break
}
fmt.Printf("处理任务: %d\n", task)
// 模拟任务处理
}
}
func main() {
var wg sync.WaitGroup
workQueues := []*WorkQueue{
&WorkQueue{},
&WorkQueue{},
&WorkQueue{},
}
// 向队列中添加任务
for i := 0; i < 10; i++ {
workQueues[i%3].AddTask(i)
}
// 启动三个worker
for i := 0; i < 3; i++ {
wg.Add(1)
go worker(workQueues[i], &wg)
}
// 等待所有worker完成
wg.Wait()
}

2.2.2.2 代码解释

• WorkQueue结构体定义了一个工作队列，包含一个任务列表和互斥锁，用于线程安全地添加和获取任务。
• AddTask方法用于向队列添加任务。
• GetTask方法用于从队列获取任务，如果队列为空，则返回false。
• worker函数是一个工作goroutine，它从自己的工作队列中获取任务，如果队列为空，则结束。
• 在main函数中，创建了三个工作队列，并向它们添加了任务。然后启动了三个worker来处理这些任务。

2.3 并发模式：扇入扇出模式

2.3.1 原理

扇入扇出模式是并发编程中用于数据流处理的一种模式。扇入是指多个生产者将数据发送到一个或多个channel，而扇出是指一个或多个消费者从这些channel接收数据并处理。这种模式可以实现数据的高效处理和传输，特别是在处理大量数据时。

2.3.2 内容

2.3.2.1 示例代码

package main
import (
"fmt"
"sync"
)
// 定义一个channel用于扇入
var inCh = make(chan int, 10)
// 定义一个channel用于扇出
var outCh = make(chan int, 10)
// 生产者函数，生成数据并发送到扇入channel
func producer(wg *sync.WaitGroup, id int) {
defer wg.Done()
for i := 0; i < 10; i++ {
inCh <- i + id*10
fmt.Printf("生产者%d生成数据: %d\n", id, i+id*10)
}
}
// 处理函数，从扇入channel接收数据，处理后发送到扇出channel
func processor(wg *sync.WaitGroup) {
defer wg.Done()
for {
select {
case data, ok := <-inCh:
if !ok {
fmt.Println("处理器接收数据完成")
return
}
// 模拟数据处理
outCh <- data * 2
}
}
}
// 消费者函数，从扇出channel接收数据并处理
func consumer(wg *sync.WaitGroup) {
defer wg.Done()
for {
select {
case data, ok := <-outCh:
if !ok {
fmt.Println("消费者接收数据完成")
return
}
fmt.Printf("消费者处理数据: %d\n", data)
}
}
}
func main() {
var wg sync.WaitGroup
// 启动两个生产者
for i := 0; i < 2; i++ {
wg.Add(1)
go producer(&wg, i)
}
// 启动处理器
wg.Add(1)
go processor(&wg)
// 启动消费者
wg.Add(1)
go consumer(&wg)
// 等待所有goroutine完成
wg.Wait()
close(inCh)
close(outCh)
}

2.3.2.2 代码解释

• inCh和outCh是用于扇入和扇出的channel。
• producer函数是生产者，它生成数据并发送到inCh。
• processor函数是处理器，它从inCh接收数据，处理后（在这个例子中是乘以2）发送到outCh。
• consumer函数是消费者，它从outCh接收数据并处理。
• 在main函数中，启动了两个生产者、一个处理器和一个消费者。通过sync.WaitGroup确保所有goroutine完成后再结束main函数。最后，关闭inCh和outCh以通知所有goroutine完成数据处理。

3 Go语言并行编程实战

3.1 网络爬虫并行抓取示例

在Go语言中，利用goroutines和channels可以高效地实现网络爬虫的并行抓取。下面的示例代码展示了如何使用这些特性来并行抓取多个网页。

package main
import (
"fmt"
"io/ioutil"
"net/http"
"sync"
)
// FetchHTML 是一个函数，用于抓取单个网页的HTML内容。
func FetchHTML(url string, wg *sync.WaitGroup, ch chan<- string) {
defer wg.Done()
resp, err := http.Get(url)
if err != nil {
ch <- fmt.Sprintf("Error fetching %s: %v", url, err)
return
}
defer resp.Body.Close()
body, err := ioutil.ReadAll(resp.Body)
if err != nil {
ch <- fmt.Sprintf("Error reading body of %s: %v", url, err)
return
}
ch <- fmt.Sprintf("Fetched %s: %d bytes", url, len(body))
}
func main() {
var wg sync.WaitGroup
ch := make(chan string)
// 假设我们有以下URL列表。
urls := []string{
"http://example.com",
"http://example.org",
"http://example.net",
}
// 为每个URL启动一个goroutine。
for _, url := range urls {
wg.Add(1)
go FetchHTML(url, &wg, ch)
}
// 等待所有goroutines完成。
go func() {
wg.Wait()
close(ch)
}()
// 从channel中读取结果。
for result := range ch {
fmt.Println(result)
}
}

3.1.1 代码解释

1. goroutines和channels的使用：我们为每个URL创建一个goroutine，这些goroutine通过channel ch 通信，将抓取结果或错误信息发送回main函数。
2. WaitGroup的使用：sync.WaitGroup 用于同步goroutines，确保所有抓取任务完成后再关闭channel。
3. 错误处理：在抓取过程中，我们检查了HTTP请求和读取响应体的错误，并通过channel发送错误信息。

3.2 并行文件系统操作示例

Go语言的并行特性同样适用于文件系统操作，如并行读取多个文件。下面的示例展示了如何并行读取多个文件的内容。

package main
import (
"fmt"
"io/ioutil"
"sync"
)
// ReadFile 是一个函数，用于并行读取文件内容。
func ReadFile(filename string, wg *sync.WaitGroup, ch chan<- string) {
defer wg.Done()
content, err := ioutil.ReadFile(filename)
if err != nil {
ch <- fmt.Sprintf("Error reading %s: %v", filename, err)
return
}
ch <- fmt.Sprintf("Read %s: %s", filename, content)
}
func main() {
var wg sync.WaitGroup
ch := make(chan string)
// 假设我们有以下文件列表。
filenames := []string{
"file1.txt",
"file2.txt",
"file3.txt",
}
// 为每个文件启动一个goroutine。
for _, filename := range filenames {
wg.Add(1)
go ReadFile(filename, &wg, ch)
}
// 等待所有goroutines完成。
go func() {
wg.Wait()
close(ch)
}()
// 从channel中读取结果。
for result := range ch {
fmt.Println(result)
}
}

3.2.1 代码解释

1. 并行读取文件：我们为每个文件创建一个goroutine，这些goroutine并行执行文件读取操作。
2. 结果收集：通过channel收集所有文件读取的结果或错误信息。
3. 同步机制：使用sync.WaitGroup确保所有文件读取操作完成后再关闭channel。

3.3 并行数据库查询示例

在Go中，处理数据库查询的并行操作可以显著提高数据处理的效率。下面的示例展示了如何并行执行多个数据库查询。

package main
import (
"database/sql"
"fmt"
"sync"
_ "github.com/go-sql-driver/mysql"
)
// QueryDB 是一个函数，用于并行执行数据库查询。
func QueryDB(db *sql.DB, query string, wg *sync.WaitGroup, ch chan<- string) {
defer wg.Done()
rows, err := db.Query(query)
if err != nil {
ch <- fmt.Sprintf("Error querying database: %v", err)
return
}
defer rows.Close()
var result string
for rows.Next() {
var field string
err := rows.Scan(&field)
if err != nil {
ch <- fmt.Sprintf("Error scanning row: %v", err)
return
}
result += field + "\n"
}
ch <- fmt.Sprintf("Query result: %s", result)
}
func main() {
var wg sync.WaitGroup
ch := make(chan string)
// 假设我们有以下数据库连接信息。
db, err := sql.Open("mysql", "user:password@tcp(127.0.0.1:3306)/dbname")
if err != nil {
fmt.Println("Error connecting to database:", err)
return
}
defer db.Close()
// 假设我们有以下查询列表。
queries := []string{
"SELECT * FROM table1",
"SELECT * FROM table2",
"SELECT * FROM table3",
}
// 为每个查询启动一个goroutine。
for _, query := range queries {
wg.Add(1)
go QueryDB(db, query, &wg, ch)
}
// 等待所有goroutines完成。
go func() {
wg.Wait()
close(ch)
}()
// 从channel中读取结果。
for result := range ch {
fmt.Println(result)
}
}

3.3.1 代码解释

1. 数据库连接：使用database/sql包和MySQL驱动程序连接到数据库。
2. 并行查询：为每个查询创建一个goroutine，这些goroutine并行执行数据库查询。
3. 结果收集：通过channel收集所有查询的结果或错误信息。
4. 错误处理：在查询和扫描结果时，我们检查了可能发生的错误，并通过channel发送错误信息。

以上示例展示了Go语言如何利用goroutines和channels实现网络抓取、文件读取和数据库查询的并行操作，从而提高程序的执行效率。

4 Go语言并行编程优化与调试

4.1 性能分析工具：pprof的使用

在Go语言中，pprof是一个强大的性能分析工具，用于收集和分析程序的性能数据，包括CPU使用率、内存使用情况、线程使用情况等。通过pprof，开发者可以识别程序中的性能瓶颈，优化并行代码的执行效率。

4.1.1 使用方法

1. 生成性能数据：在程序中使用pprof包的Profile函数来收集性能数据。

import (
"net/http"
"net/http/pprof"
)
func main() {
http.ListenAndServe(":8080", nil)
http.HandleFunc("/debug/pprof/", pprof.Index)
http.HandleFunc("/debug/pprof/cmdline", pprof.Cmdline)
http.HandleFunc("/debug/pprof/profile", pprof.Profile)
http.HandleFunc("/debug/pprof/symbol", pprof.Symbol)
http.HandleFunc("/debug/pprof/trace", pprof.Trace)
}

1. 分析性能数据：使用pprof命令行工具分析生成的性能数据。

• go tool pprof http://localhost:8080/debug/pprof/profile

4.1.2 示例

假设我们有一个Go程序，其中包含一个并行处理大量数据的函数。我们可以使用pprof来分析这个函数的性能。

package main
import (
"fmt"
"net/http"
"net/http/pprof"
"sync"
"time"
)
func processData(data []int, wg *sync.WaitGroup) {
defer wg.Done()
for _, d := range data {
time.Sleep(time.Millisecond * 100) // 模拟耗时操作
fmt.Println("Processed:", d)
}
}
func main() {
data := make([]int, 1000)
for i := range data {
data[i] = i
}
var wg sync.WaitGroup
for i := 0; i < 10; i++ {
wg.Add(1)
go processData(data, &wg)
}
wg.Wait()
http.ListenAndServe(":8080", nil)
http.HandleFunc("/debug/pprof/", pprof.Index)
http.HandleFunc("/debug/pprof/cmdline", pprof.Cmdline)
http.HandleFunc("/debug/pprof/profile", pprof.Profile)
http.HandleFunc("/debug/pprof/symbol", pprof.Symbol)
http.HandleFunc("/debug/pprof/trace", pprof.Trace)
}

通过访问http://localhost:8080/debug/pprof/profile，我们可以下载到性能数据文件，然后使用go tool pprof命令进行分析，找出哪些函数或代码段消耗了最多的CPU时间。

4.2 避免死锁和竞态条件

在并行编程中，死锁和竞态条件是常见的问题。死锁发生在两个或多个goroutine相互等待对方释放资源时，而竞态条件则发生在多个goroutine同时访问共享资源，导致数据不一致或程序行为异常。

4.2.1 避免死锁

1. 避免循环等待：确保资源的获取顺序一致，避免形成循环等待的链。
2. 使用超时：在可能的情况下，为资源获取操作设置超时，以避免无限等待。
3. 使用channel进行同步：channel可以作为goroutine之间的通信和同步机制，避免死锁。

4.2.2 避免竞态条件

1. 使用互斥锁：通过sync.Mutex来保护共享资源，确保同一时间只有一个goroutine可以访问。
2. 使用原子操作：对于简单的计数器或状态变量，可以使用sync/atomic包中的原子操作来避免竞态条件。
3. 使用context包：context包提供了取消操作的功能，可以用来控制goroutine的生命周期，避免竞态条件。

4.2.3 示例

下面的代码示例展示了如何使用sync.Mutex来避免竞态条件：

package main
import (
"fmt"
"sync"
"time"
)
var count int
var mutex sync.Mutex
func increment() {
mutex.Lock()
count++
mutex.Unlock()
}
func main() {
var wg sync.WaitGroup
for i := 0; i < 1000; i++ {
wg.Add(1)
go func() {
defer wg.Done()
increment()
}()
}
wg.Wait()
fmt.Println("Final count:", count)
}

在这个例子中，我们使用sync.Mutex来保护count变量，确保在并行执行时，count的值不会因为竞态条件而出现错误。

4.3 并行代码的测试与调试

测试并行代码比测试顺序代码更具挑战性，因为并行代码的行为可能依赖于执行的顺序，这在不同的运行环境中可能会有所不同。

4.3.1 测试策略

1. 使用race检测器：Go的race检测器可以帮助识别竞态条件。
2. 编写并行测试用例：确保测试用例能够模拟并行执行的场景。
3. 使用testing包的T.Parallel函数：允许测试用例并行执行，以检测潜在的竞态条件。

4.3.2 示例

下面的代码示例展示了如何使用race检测器和T.Parallel函数来测试并行代码：

package main
import (
"fmt"
"sync"
"testing"
)
var count int
var mutex sync.Mutex
func increment(t *testing.T) {
mutex.Lock()
count++
mutex.Unlock()
t.Log("Incremented count:", count)
}
func TestIncrement(t *testing.T) {
t.Run("Parallel", func(t *testing.T) {
t.Parallel()
for i := 0; i < 1000; i++ {
go increment(t)
}
time.Sleep(time.Second)
if count != 1000 {
t.Errorf("Expected count to be 1000, got %d", count)
}
})
}

在这个测试用例中，我们使用T.Parallel函数来并行执行increment函数，同时使用race检测器来检查是否有竞态条件发生。

4.3.3 调试技巧

1. 使用debug/pprof：通过debug/pprof可以获取程序的执行状态，包括goroutine的状态和堆栈信息。
2. 使用日志：在关键代码段添加日志输出，可以帮助理解程序的执行流程和状态。
3. 使用testing包的T.Log函数：在测试用例中使用T.Log函数输出调试信息，可以帮助定位问题。

通过这些测试和调试策略，我们可以更有效地识别和解决并行代码中的问题，提高代码的稳定性和性能。

5 Go语言并行编程高级话题

5.1 context包在并发中的应用

在Go语言中，context包提供了一种在并发程序中传递取消信号、截止时间以及其它请求特定的值的方式。这在处理长时间运行的请求或在多个goroutine之间协调时特别有用。

5.1.1 原理

context包的核心是Context接口，它定义了几个方法，包括Deadline(), Done(), Err(), 和 Value(key interface{}) interface{}。context包还提供了创建Context的几种方式，如Background(), TODO(), WithCancel(), WithDeadline(), WithTimeout(), 和 WithValue()。

5.1.2 内容

5.1.2.1 创建Context

• context.Background()：创建一个没有截止时间的Context，通常用于顶层调用。
• context.TODO()：与Background()类似，但表示没有设置Context的意图。
• context.WithCancel(parent Context)：创建一个可以被取消的子Context。
• context.WithDeadline(parent Context, deadline time.Time)：创建一个在指定时间后会自动取消的子Context。
• context.WithTimeout(parent Context, timeout time.Duration)：创建一个在给定时间后会自动取消的子Context。
• context.WithValue(parent Context, key, val interface{})：创建一个带有特定值的子Context，用于传递请求特定的值。

5.1.2.2 使用示例

package main
import (
"context"
"fmt"
"time"
)
func main() {
// 创建一个带有截止时间的Context
ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
defer cancel()
// 使用Context来控制goroutine的运行
go func() {
select {
case <-ctx.Done():
fmt.Println("任务被取消")
}
}()
// 等待一段时间后取消Context
time.Sleep(500 * time.Millisecond)
cancel()
time.Sleep(500 * time.Millisecond) // 确保goroutine已经接收到取消信号
}

在这个例子中，我们创建了一个带有1秒截止时间的Context。然后，我们启动了一个goroutine，它会监听ctx.Done()通道。如果在500毫秒后我们调用cancel()函数，goroutine会在接下来的500毫秒内接收到取消信号，并打印出“任务被取消”。

5.2 Go的并行调度器原理

Go语言的并行调度器是其并发模型的核心，它负责管理goroutine的执行和切换。调度器的设计目标是高效地利用多核处理器，同时保持程序的简单性和可预测性。

5.2.1 原理

Go的调度器使用了工作窃取（Work Stealing）算法，这意味着goroutine可以在不同的CPU核心上执行，即使它们属于同一个线程。调度器维护了一个全局的goroutine队列和每个线程的本地队列。当一个线程的本地队列为空时，它可以从全局队列中窃取任务来执行，或者从其他线程的本地队列中窃取。

5.2.2 内容

5.2.2.1 调度器的关键组件

• M：代表机器，即运行goroutine的线程。
• P：代表处理器，每个P都有一个本地队列，用于存储待执行的goroutine。
• G：代表goroutine，是Go语言中的轻量级线程。

5.2.2.2 调度器的工作流程

1. 创建goroutine：当一个goroutine被创建时，它会被添加到全局队列或某个P的本地队列中。
2. 调度goroutine：调度器会根据CPU核心的数量和当前的goroutine队列情况，决定将goroutine分配给哪个M执行。
3. 工作窃取：当一个M的本地队列为空时，它会尝试从全局队列或其它M的本地队列中窃取goroutine来执行。
4. goroutine切换：当一个goroutine执行到I/O操作、系统调用或主动调用runtime.Gosched()时，调度器会切换到另一个goroutine执行。

5.3 并行编程中的错误处理机制

在并行编程中，错误处理变得尤为重要，因为错误可能在任何goroutine中发生。Go语言提供了一种优雅的方式来处理并发中的错误，这通常涉及到使用error类型和context包。

5.3.1 原理

在Go中，错误通常通过函数返回值来传递。当一个函数可能失败时，它通常会返回一个error类型的值。在并发场景中，错误可以通过context包的Err()方法来传递，这允许在多个goroutine之间传播错误。

5.3.2 内容

5.3.2.1 错误处理的最佳实践

• 使用error类型：确保所有可能失败的函数都返回一个error类型的值。
• 检查错误：在调用函数后立即检查返回的错误，不要忽略它。
• 使用context取消操作：当一个操作需要在多个goroutine中执行时，使用context来协调它们的执行和错误处理。
• 错误聚合：在处理多个并发操作时，使用sync.WaitGroup和errors包来聚合所有可能的错误。

5.3.2.2 使用示例

package main
import (
"context"
"errors"
"fmt"
"sync"
)
func fetchData(ctx context.Context, wg *sync.WaitGroup, id int) error {
defer wg.Done()
select {
case <-ctx.Done():
return ctx.Err()
default:
// 模拟数据获取
time.Sleep(time.Duration(id) * time.Second)
if id == 3 {
return errors.New("数据获取失败")
}
fmt.Printf("数据%d获取成功\n", id)
return nil
}
}
func main() {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
var wg sync.WaitGroup
wg.Add(3)
for i := 1; i <= 3; i++ {
go fetchData(ctx, &wg, i)
}
wg.Wait()
}

在这个例子中，我们创建了一个带有5秒截止时间的Context。然后，我们启动了3个goroutine来获取数据。如果数据获取操作超过了5秒，fetchData函数会接收到取消信号，并返回一个错误。如果数据获取失败，它也会返回一个错误。最后，我们使用sync.WaitGroup来等待所有goroutine完成，这有助于我们聚合所有可能的错误。