[每周一更]-(第131期)：Go并发协程总结篇

news2025/1/13 11:23:00

在这里插入图片描述

Go语言的并发是通过协程（goroutine）实现的。Go协程是轻量级的线程，允许多个任务同时执行，且Go运行时会高效地管理它们。在Go中使用并发协程的方式非常简便，也很强大。以下是一些关于Go协程的基础用法和并发控制方法：

文章目录

- - 1. 启动协程
  - 2. 使用 `sync.WaitGroup` 管理协程
  - 3. 使用通道（Channel）进行协程间通信
  - - 创建并使用通道
    - 带缓冲的通道
  - 4. 使用 `select` 多路复用通道
  - 5. 使用 `sync.Mutex` 实现互斥锁
  - 6. 使用 `sync.Once` 确保只执行一次
  - - 小结
  - 7.用 `context` 管理并发任务
  - - 实际业务场景

1. 启动协程

要启动一个协程，只需要在函数调用前加上关键字 go。

package main

import (
    "fmt"
    "time"
)

func printMessage(message string) {
    fmt.Println(message)
}

func main() {
    go printMessage("Hello from goroutine!")
    fmt.Println("Hello from main!")
    time.Sleep(time.Second) // 给协程时间执行完毕
}

在这个例子中，printMessage 函数会作为一个协程执行，因此main函数中的打印语句和协程可能会并行执行。

注意：main函数运行完毕时，程序会立即退出，即使其他协程仍在执行。所以在这里用time.Sleep让主协程等待，确保子协程有时间完成工作。

2. 使用 `sync.WaitGroup` 管理协程

sync.WaitGroup 是Go提供的一个结构，可以用来等待一组协程完成。

package main

import (
    "fmt"
    "sync"
)

func printNumber(num int, wg *sync.WaitGroup) {
    defer wg.Done() // 调用Done来减少计数
    fmt.Println(num)
}

func main() {
    var wg sync.WaitGroup
    for i := 1; i <= 5; i++ {
        wg.Add(1) // 增加计数
        go printNumber(i, &wg)
    }
    wg.Wait() // 等待所有协程完成
    fmt.Println("All goroutines finished!")
}

3. 使用通道（Channel）进行协程间通信

通道是Go内置的并发控制机制，协程间可以通过通道进行数据传递，避免数据竞争。

channel	nil	非空	空的	满的	没满
接收	阻塞	接收值	阻塞	接收值	接收值
发送	阻塞	发送值	发送值	阻塞	发送值
关闭	panic	关闭成功，读完数据后返回零值	关闭成功，返回零值	关闭成功，读完数据后返回零值	关闭成功，读完数据后，返回零值

关闭后的通道有以下特点：

对一个关闭的通道再发送值就会导致panic。
对一个关闭的通道进行接收会一直获取值直到通道为空。
对一个关闭的并且没有值的通道执行接收操作会得到对应类型的零值。
关闭一个已经关闭的通道会导致panic。

创建并使用通道

package main

import (
    "fmt"
)

func sum(a, b int, resultChan chan int) {
    result := a + b
    resultChan <- result // 将结果发送到通道
}

func main() {
    resultChan := make(chan int) // 创建一个int类型的通道

    go sum(3, 5, resultChan) // 启动协程计算
    result := <-resultChan   // 从通道中接收结果
    fmt.Println("Sum:", result)
}

带缓冲的通道

Go支持缓冲通道，允许发送者发送多于一个数据而不需要立刻被接收。

package main

import (
    "fmt"
)

func main() {
    bufferedChan := make(chan int, 3) // 缓冲大小为3
    bufferedChan <- 1
    bufferedChan <- 2
    bufferedChan <- 3
    fmt.Println(<-bufferedChan)
    fmt.Println(<-bufferedChan)
    fmt.Println(<-bufferedChan)
}

4. 使用 `select` 多路复用通道

select 语句可以同时等待多个通道操作，执行第一个准备好的操作。

package main

import (
    "fmt"
    "time"
)

func main() {
    chan1 := make(chan string)
    chan2 := make(chan string)

    go func() {
        time.Sleep(2 * time.Second)
        chan1 <- "Message from channel 1"
    }()
    go func() {
        time.Sleep(1 * time.Second)
        chan2 <- "Message from channel 2"
    }()

    select {
    case msg1 := <-chan1:
        fmt.Println(msg1)
    case msg2 := <-chan2:
        fmt.Println(msg2)
    case <-time.After(3 * time.Second):
        fmt.Println("Timeout")
    }
}

5. 使用 `sync.Mutex` 实现互斥锁

当多个协程需要访问共享资源时，可以使用sync.Mutex来保证同一时间只有一个协程可以访问。

package main

import (
    "fmt"
    "sync"
)

var counter int
var mutex sync.Mutex

func increment(wg *sync.WaitGroup) {
    defer wg.Done()
    mutex.Lock()   // 加锁
    counter++
    mutex.Unlock() // 解锁
}

func main() {
    var wg sync.WaitGroup

    for i := 0; i < 10; i++ {
        wg.Add(1)
        go increment(&wg)
    }

    wg.Wait()
    fmt.Println("Final counter:", counter)
}

6. 使用 `sync.Once` 确保只执行一次

有些场景下，需要确保某段代码只执行一次（例如初始化配置），可以使用sync.Once。

package main

import (
    "fmt"
    "sync"
)

var once sync.Once

func initialize() {
    fmt.Println("Initialized!")
}

func main() {
    for i := 0; i < 3; i++ {
        once.Do(initialize) // 只会执行一次
    }
}

小结

启动协程：使用go关键字。
等待协程完成：使用sync.WaitGroup。
协程间通信：使用通道（Channel）。
通道多路复用：使用select。
互斥控制：使用sync.Mutex。
单次执行：使用sync.Once。

Go的并发特性为编写高效的并发程序提供了强大支持，同时也有丰富的同步原语来避免竞态条件。

7.用 `context` 管理并发任务

使用 context.Context 可以优雅地管理并发任务，尤其是在需要统一控制超时、取消操作或共享上下文数据时。

实际业务场景

超时管理：
- API 请求或任务需要在一定时间内完成，否则强制取消，避免资源浪费。
任务协调：
- 多个并发任务共享相同的上下文状态，例如取消操作。
优雅退出：
- 确保并发任务能够响应外部取消信号，安全退出而非强制终止。

package main

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

// 模拟一个工作函数
func worker(ctx context.Context, id int, wg *sync.WaitGroup) {
	defer wg.Done()

	for {
		select {
		case <-ctx.Done():
			// 任务被取消时结束
			fmt.Printf("Worker %d stopped: %v\n", id, ctx.Err())
			return
		default:
			// 模拟工作
			time.Sleep(time.Duration(rand.Intn(100)) * time.Millisecond)
			fmt.Printf("Worker %d is processing\n", id)
		}
	}
}

func main() {
	// 设置超时时间的上下文
	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()

	// 创建 WaitGroup 来等待所有任务完成
	var wg sync.WaitGroup

	// 启动多个并发任务
	numWorkers := 5
	for i := 1; i <= numWorkers; i++ {
		wg.Add(1)
		go worker(ctx, i, &wg)
	}

	// 等待所有任务结束
	wg.Wait()
	fmt.Println("All workers stopped.")
}