Golang 并发机制-5：详解syn包同步原语

并发性是现代软件开发的一个基本方面，Go（也称为Golang）为并发编程提供了一组健壮的工具。Go语言中用于管理并发性的重要包之一是“sync”包。在本文中，我们将概述“sync”包，并深入研究其最重要的同步原语之一：Wait
Groups.

sync 包概述

sync包是Go中的一个标准库包，为并发编程提供同步原语。它为开发人员提供了协调和同步程序的工具，确保安全有序地执行并发任务。sync包提供的一些关键同步原语包括Mutexes, RWMutexes, Cond, Wait Groups。

Wait Groups

Wait Group是由Go中的“sync”包提供的同步原语。它是一个简单但功能强大的工具，用于管理goroutine的同步，特别是当你希望等待一组goroutine在继续之前完成它们的任务时。

当有多个并发执行独立任务的goroutine，并且你需要确保它们在继续执行主程序之前都已完成执行时，等待组是有用的。
在这里插入图片描述

让我们通过一个代码示例来探索如何使用Wait Groups：

package main

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

func worker(id int, wg *sync.WaitGroup) {
	defer wg.Done() // Decrement the Wait Group counter when done
	fmt.Printf("Worker %d is working\n", id)
	time.Sleep(time.Second)
	fmt.Printf("Worker %d has finished\n", id)
}

func main() {
	var wg sync.WaitGroup

	for i := 1; i <= 3; i++ {
		wg.Add(1) // Increment the Wait Group counter for each Goroutine
		go worker(i, &wg)
	}

	wg.Wait() // Wait for all Goroutines to finish
	fmt.Println("All workers have finished.")
}

在这个例子中，我们定义了一个worker函数，它通过睡眠一秒钟来模拟工作。我们启动三个goroutine，每个代表一个worker，并使用sync。来协调他们的执行。

wg.Add(1)在启动每个例程之前增加等待组计数器。
wg.Done()在worker函数中被延迟，以在gooutine完成其工作时减少计数器。
wg.Wait()阻塞主程序，直到所有的例程都完成，确保我们等待所有工人的完成。

RWMutex

RWMutex（读写互斥）是Go语言中的一个同步原语，它允许多个线程同时读取共享数据，同时确保对写入的独占访问。它在经常读取数据但不经常修改数据的场景中非常有用。

下面是一个演示如何使用RWMutex的简单示例：

package main

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

var (
	data        int
	dataMutex   sync.RWMutex
)

func readData() int {
	dataMutex.RLock() // Read Lock
	defer dataMutex.RUnlock()
	return data
}

func writeData(value int) {
	dataMutex.Lock() // Write Lock
	defer dataMutex.Unlock()
	data = value
}

func main() {
	// Read data concurrently
	for i := 1; i <= 5; i++ {
		go func() {
			fmt.Println("Read Data:", readData())
		}()
	}

	// Write data
	writeData(42)

	time.Sleep(time.Second)
}

在这个例子中，多个Goroutine并发地读取共享的“数据”，并且一个单独的Goroutine写入它。RWMutex确保多个读取器可以同时访问数据，但一次只有一个写入器可以修改数据。

什么是条件变量？

在 Go 语言里，条件变量（sync.Cond）是一种同步原语，它用于协调多个 goroutine 的执行顺序，尤其是在某个条件满足时唤醒等待的 goroutine。条件变量通常和互斥锁（sync.Mutex 或 sync.RWMutex）一起使用，互斥锁用于保护共享资源，而条件变量则用于在共享资源的状态发生变化时通知等待的 goroutine。

主要特性

1. 等待（Wait）

Cond.Wait() 方法会让当前 goroutine 进入等待状态，并且会自动释放与之关联的互斥锁。当其他 goroutine 调用 Cond.Signal() 或 Cond.Broadcast() 唤醒它时，该 goroutine 会重新获取互斥锁并继续执行。

2. 单发通知（Signal）

Cond.Signal() 方法会唤醒一个正在等待该条件变量的 goroutine。如果有多个 goroutine 在等待，它会选择其中一个进行唤醒。

3. 广播通知（Broadcast）

Cond.Broadcast() 方法会唤醒所有正在等待该条件变量的 goroutine。

条件变量（Condition Variables）是同步原语，它允许程序在继续之前等待特定条件变为真。当你需要根据特定条件协调多个goroutine的执行时，它们很有帮助。

下面是一个说明条件变量使用的基本示例：

package main

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

var (
	conditionMutex sync.Mutex
	condition      *sync.Cond
	isReady        bool
)

func waitForCondition() {
	conditionMutex.Lock()
	defer conditionMutex.Unlock()

	for !isReady {
		fmt.Println("Waiting for the condition...")
		condition.Wait()
	}
	fmt.Println("Condition met, proceeding.")
}

func setCondition() {
	time.Sleep(2 * time.Second)
	conditionMutex.Lock()
	isReady = true
	condition.Signal() // Signal one waiting Goroutine
	conditionMutex.Unlock()
}

func main() {
	condition = sync.NewCond(&conditionMutex)

	go waitForCondition()
	go setCondition()

	time.Sleep(5 * time.Second)
}

在这个例子中，一个Goroutine使用condition. wait()等待条件变为真，而另一个Goroutine将条件设置为true，并使用condition. signal()向等待的Goroutine发出信号。

下面是一个简单的示例，模拟生产者 - 消费者模型，使用条件变量来协调生产者和消费者的行为：

package main

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

// 定义一个缓冲区和相关的锁与条件变量
var (
    buffer    []int
    bufferLen = 5
    mutex     sync.Mutex
    cond      = sync.NewCond(&mutex)
)

// 生产者函数
func producer(id int) {
    for {
        mutex.Lock()
        // 检查缓冲区是否已满
        for len(buffer) == bufferLen {
            fmt.Printf("Producer %d is waiting as buffer is full...\n", id)
            cond.Wait() // 缓冲区满，等待消费者消费
        }
        // 生产一个元素
        item := len(buffer) + 1
        buffer = append(buffer, item)
        fmt.Printf("Producer %d produced item %d. Buffer: %v\n", id, item, buffer)
        cond.Signal() // 通知可能正在等待的消费者
        mutex.Unlock()
        time.Sleep(time.Second)
    }
}

// 消费者函数
func consumer(id int) {
    for {
        mutex.Lock()
        // 检查缓冲区是否为空
        for len(buffer) == 0 {
            fmt.Printf("Consumer %d is waiting as buffer is empty...\n", id)
            cond.Wait() // 缓冲区空，等待生产者生产
        }
        // 消费一个元素
        item := buffer[0]
        buffer = buffer[1:]
        fmt.Printf("Consumer %d consumed item %d. Buffer: %v\n", id, item, buffer)
        cond.Signal() // 通知可能正在等待的生产者
        mutex.Unlock()
        time.Sleep(time.Second)
    }
}

func main() {
    // 启动生产者和消费者 goroutine
    go producer(1)
    go consumer(1)

    // 让程序运行一段时间
    time.Sleep(10 * time.Second)
}

通过使用条件变量，生产者和消费者能够在合适的时机进行等待和唤醒，确保缓冲区不会溢出或空消费。

原子操作

原子操作（Atomic Operations ）是作为单个、不可分割的工作单元执行的操作。它们通常用于在不需要互斥锁的情况下安全地更新并发程序中的共享变量。Go为原子操作提供了一个名为“atomic”的包。

下面是演示原子操作的例子：

package main

import (
	"fmt"
	"sync"
	"sync/atomic"
	"time"
)

var (
	counter int32
	wg      sync.WaitGroup
)

func incrementCounter() {
	defer wg.Done()
	for i := 0; i < 100000; i++ {
		atomic.AddInt32(&counter, 1)
	}
}

func main() {
	wg.Add(2)
	go incrementCounter()
	go incrementCounter()
	wg.Wait()

	fmt.Println("Counter:", atomic.LoadInt32(&counter))
}