理解Go中的syncCond：生产者-消费者场景中的Goroutine同步

小羊标书

一键生成百页标书，让投标更简单高效

下载

在并发编程中，同步是防止数据竞争并确保线程或 goroutine 以协调方式运行的关键。 想象一下，您在协调多个生产者和消费者访问共享资源（例如缓冲区或队列）时遇到问题。这种经典的并发挑战被称为生产者-消费者问题。在这种情况下，同步对于确保生产者不会覆盖数据以及消费者不会读取无效或陈旧的数据至关重要。同步是必要的，因为如果没有正确的同步，同时访问共享数据可能会导致竞争条件、数据损坏或崩溃。如果缓冲区已满，生产者需要等待，如果缓冲区为空，消费者需要等待。 在某些情况下您有一个固定大小的有界缓冲区，并且您需要管理多个生产者和消费者之间对其的访问。

• 阻塞：goroutines 可以等待信号，暂停执行直到收到通知。
• 发信号：当满足条件时，其他 goroutine 可以向等待的 goroutine 发出信号以继续执行。
• 效率：通过让 goroutine 休眠直到收到信号来减少繁忙等待。

• sync.cond初始化：它需要一个locker，通常是sync.mutex或sync.rwmutex来控制访问。此储物柜有助于保护共享资源。
• wait()：当 goroutine 调用 wait() 时，它：
  释放关联的锁，允许其他 goroutine 访问该资源。
  等待（阻塞）直到另一个 goroutine 发出继续的信号。
• 信号（）和广播（）：
  • signal() 唤醒一个 等待的 goroutine，允许其获取锁并继续。
  • broadcast() 唤醒所有 等待的 goroutine。

缓冲区（或队列），其大小固定。多个生产者生成项目并将其添加到缓冲区，而多个消费者则从中删除项目。挑战是：

确保生产者仅在缓冲区有空间时添加项目。
确保消费者仅在缓冲区不为空时移除项目。
向生产者和消费者发出可以添加或删除项目的信号。

package main

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

const buffersize = 5

type buffer struct {
    data []int
    mu   sync.mutex
    cond *sync.cond
}

func (b *buffer) produce(item int) {
    // producer logic to add item to the buffer
}

func (b *buffer) consume() int {
    // consumer logic to remove item from the buffer
    return 0
}

func main() {
    buffer := &buffer{data: make([]int, 0, buffersize)}
    buffer.cond = sync.newcond(&buffer.mu)
    var wg sync.waitgroup

    // start producer goroutines
    for i := 1; i <= 3; i++ {
        wg.add(1)
        go func(id int) {
            defer wg.done()
            for j := 0; j < 5; j++ { // each producer creates 5 items
                buffer.produce(id*10 + j) // produce unique items based on id and j
                time.sleep(100 * time.millisecond)
            }
        }(i)
    }

    // start consumer goroutines
    for i := 1; i <= 3; i++ {
        wg.add(1)
        go func(id int) {
            defer wg.done()
            for j := 0; j < 5; j++ { // each consumer consumes 5 items
                item := buffer.consume()
                fmt.printf("consumer %d consumed item %d\n", id, item)
                time.sleep(150 * time.millisecond)
            }
        }(i)
    }

    wg.wait()
    fmt.println("all producers and consumers finished.")
}

初始化:

buffer.cond = sync.newcond(&buffer.mu)

这里，sync.newcond(&buffer.mu) 创建一个与 mu 互斥体关联的新条件变量。条件变量可以等待缓冲区的更改并发出信号（例如添加或删除项目）。

生产者方法（生产）:

func (b *buffer) produce(item int) {
    b.mu.lock()
    defer b.mu.unlock()

    // wait if the buffer is full
    for len(b.data) == buffersize {
        b.cond.wait() // release lock and wait until signaled
    }

    // add item to the buffer
    b.data = append(b.data, item)
    fmt.printf("produced item %d\n", item)

    // signal a consumer that an item is available
    b.cond.signal()
}

• 锁定：生产者锁定 mu，以确保其对 b.data 具有独占访问权限。
• wait if full：如果缓冲区已满，生产者调用 b.cond.wait()：
  这会释放 b.mu 上的锁定，允许消费者使用缓冲区中的项目。
  它会等待（阻塞），直到消费者发出信号表明缓冲区中现在有空间。
• 添加项目和信号：一旦缓冲区中有空间，生产者：
  将项目添加到缓冲区。
  调用 b.cond.signal() 通知一个等待的消费者（如果有）现在有一个项目可以消费。

消费方法（消费）:

func (b *Buffer) consume() int {
    b.mu.Lock()
    defer b.mu.Unlock()

    // Wait if the buffer is empty
    for len(b.data) == 0 {
        b.cond.Wait() // Release lock and wait until signaled
    }

    // Remove item from the buffer
    item := b.data[0]
    b.data = b.data[1:]
    fmt.Printf("Consumed item %d\n", item)

    // Signal a producer that space is available
    b.cond.Signal()

    return item
}

• 锁定：消费者锁定 mu 以确保对 b.data 的独占访问。
• wait if empty：如果缓冲区为空，消费者调用 b.cond.wait()：
  这会释放 b.mu 上的锁定，允许生产者生产物品并在准备就绪时发出信号。
  消费者等待直到有商品可供消费。
• 消费项目和信号：一旦缓冲区中有一个项目，消费者：
  删除它。
  调用 b.cond.signal() 通知等待的生产者缓冲区中现在有空间。

• 条件变量：sync.cond 提供了一种有效的方法来处理缓冲区已满或空的情况，而无需进行不必要的循环。
• 等待和信号机制：wait() 自动释放锁，这可以通过允许其他 goroutine 在适当的时候继续执行来防止死锁。
• 协调：通过使用 signal()，我们协调生产者和消费者的操作，确保每个人仅在必要时等待，防止它们在空或满的缓冲区上进行操作。

生产者
• 等待如果缓冲区已满，并在生产完产品后向消费者发出信号。
消费者
• 等待如果缓冲区为空，并且在消费完物品后向生产者发出信号。

想象一下您的任务多个 goroutine 需要等待特定条件才能继续，例如：

• 批处理：等待任务积累到一定数量才一起处理。
• 事件协调：等待事件发生（例如，要加载的数据、可用的资源）。
• rate limiting：控制并发操作数量，防止资源耗尽。 在这些场景中，sync.cond 提供了一种根据条件管理 goroutine 同步的有效方法，非常适合需要在并发任务之间进行协调的问题。