MK
摩柯社区 - 一个极简的技术知识社区
AI 面试

Goroutine并发模型与实战

2024-06-045.8k 阅读

Goroutine基础

在Go语言中，Goroutine是实现并发编程的核心机制。它类似于线程，但又有很大的区别。线程是操作系统层面的概念，而Goroutine是由Go运行时（runtime）管理的轻量级执行单元。

创建一个Goroutine非常简单，只需在函数调用前加上go关键字。例如：

package main

import (
    "fmt"
    "time"
)

func say(s string) {
    for i := 0; i < 5; i++ {
        time.Sleep(100 * time.Millisecond)
        fmt.Println(s)
    }
}

func main() {
    go say("world")
    say("hello")
}

在上述代码中，go say("world")创建了一个新的Goroutine来执行say("world")函数。同时，主函数中继续执行say("hello")。这两个函数调用是并发执行的。

Goroutine的调度模型

Go语言采用M:N的调度模型，即多个Goroutine映射到多个操作系统线程上。这种模型使得Go运行时可以高效地管理大量的Goroutine。

在Go的调度模型中，有三个重要的概念：M（操作系统线程）、P（处理器）和G（Goroutine）。

  • M：代表操作系统线程，由操作系统内核管理。每个M都有自己的栈空间。
  • P：处理器，它包含了运行Goroutine的资源，如Goroutine队列。P的数量可以通过runtime.GOMAXPROCS函数设置，默认值是CPU的核心数。
  • G：Goroutine，轻量级执行单元，每个G都有自己的栈空间和执行上下文。

Go运行时通过调度器将Goroutine分配到P上，再由P绑定到M上执行。当一个Goroutine阻塞时，调度器会将其他Goroutine调度到这个M上执行，从而提高系统的并发性能。

并发通信：Channel

虽然Goroutine提供了并发执行的能力，但如何在多个Goroutine之间进行通信和同步是一个关键问题。Go语言通过Channel来解决这个问题。

Channel是一种类型安全的管道，用于在Goroutine之间传递数据。它就像一个传送带，数据从一端发送，从另一端接收。

创建和使用Channel

创建一个Channel非常简单，使用内置的make函数：

ch := make(chan int)

上述代码创建了一个可以传递整数类型数据的Channel。

发送数据到Channel使用<-操作符：

ch <- 10

从Channel接收数据也使用<-操作符：

num := <-ch

下面是一个完整的示例：

package main

import (
    "fmt"
)

func sum(s []int, c chan int) {
    sum := 0
    for _, v := range s {
        sum += v
    }
    c <- sum
}

func main() {
    s := []int{7, 2, 8, -9, 4, 0}

    c := make(chan int)
    go sum(s[:len(s)/2], c)
    go sum(s[len(s)/2:], c)
    x, y := <-c, <-c

    fmt.Println(x, y, x+y)
}

在这个例子中，我们创建了一个Channel c，并启动两个Goroutine来分别计算切片s的前半部分和后半部分的和。最后，从Channel中接收这两个计算结果并相加。

Channel的类型

Channel有两种类型：无缓冲Channel和有缓冲Channel。

  • 无缓冲Channel：创建时没有指定缓冲区大小，如ch := make(chan int)。这种Channel要求发送操作和接收操作必须同时准备好，否则会发生阻塞。这就像是两个人在传递物品，必须一个人递出的同时另一个人伸手接住。
  • 有缓冲Channel：创建时指定了缓冲区大小，如ch := make(chan int, 5)。这种Channel可以在缓冲区未满时，发送操作不阻塞；在缓冲区不为空时，接收操作不阻塞。就好比有一个容量为5的箱子，物品可以先放在箱子里，等人来取。

同步机制：Mutex和WaitGroup

虽然Channel在很多情况下可以满足并发编程中的同步需求，但在某些场景下，我们还需要其他同步机制，比如Mutex（互斥锁）和WaitGroup。

Mutex

Mutex用于保护共享资源，防止多个Goroutine同时访问导致数据竞争。

在Go语言中，使用sync.Mutex类型来实现互斥锁。下面是一个简单的示例：

package main

import (
    "fmt"
    "sync"
)

var (
    counter int
    mu      sync.Mutex
)

func increment(wg *sync.WaitGroup) {
    defer wg.Done()
    mu.Lock()
    counter++
    mu.Unlock()
}

func main() {
    var wg sync.WaitGroup
    for i := 0; i < 1000; i++ {
        wg.Add(1)
        go increment(&wg)
    }
    wg.Wait()
    fmt.Println("Final counter value:", counter)
}

在这个例子中，counter是一个共享资源，多个Goroutine可能会同时对其进行增加操作。通过mu.Lock()和mu.Unlock()来保护counter，确保同一时间只有一个Goroutine可以访问它。

WaitGroup

WaitGroup用于等待一组Goroutine完成。它有三个主要方法：Add、Done和Wait。

  • Add方法用于添加需要等待的Goroutine数量。
  • Done方法用于标记一个Goroutine已经完成。
  • Wait方法用于阻塞当前Goroutine，直到所有添加的Goroutine都调用了Done方法。

上述示例中已经展示了WaitGroup的基本用法。wg.Add(1)表示增加一个需要等待的Goroutine，defer wg.Done()在increment函数结束时标记该Goroutine已完成，wg.Wait()则等待所有Goroutine完成。

Goroutine并发模型实战：生产者 - 消费者模型

生产者 - 消费者模型是一种常见的并发模型，在这种模型中，生产者Goroutine生成数据并将其发送到Channel，消费者Goroutine从Channel中接收数据并进行处理。

下面是一个简单的生产者 - 消费者模型的实现：

package main

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

func producer(id int, out chan<- int, wg *sync.WaitGroup) {
    defer wg.Done()
    for i := 0; i < 5; i++ {
        fmt.Printf("Producer %d sending %d\n", id, i)
        out <- i
        time.Sleep(time.Millisecond * 100)
    }
}

func consumer(id int, in <-chan int, wg *sync.WaitGroup) {
    defer wg.Done()
    for val := range in {
        fmt.Printf("Consumer %d received %d\n", id, val)
        time.Sleep(time.Millisecond * 200)
    }
}

func main() {
    var wg sync.WaitGroup
    ch := make(chan int)

    wg.Add(2)
    go producer(1, ch, &wg)
    go producer(2, ch, &wg)

    wg.Add(2)
    go consumer(1, ch, &wg)
    go consumer(2, ch, &wg)

    go func() {
        time.Sleep(time.Second)
        close(ch)
    }()

    wg.Wait()
}

在这个示例中，我们有两个生产者Goroutine和两个消费者Goroutine。生产者生成数据并发送到Channel，消费者从Channel接收数据并处理。time.Sleep模拟了实际的生产和消费过程中的耗时操作。close(ch)用于关闭Channel，当Channel关闭后，消费者的for...range循环会自动结束。

Goroutine并发模型实战：扇入（Fan - In）和扇出（Fan - Out）

扇出（Fan - Out）

扇出是指将一个任务分发给多个Goroutine去并行处理。例如，我们有一个任务是计算一个切片中每个元素的平方，为了提高效率，可以将这个任务分发给多个Goroutine。

package main

import (
    "fmt"
    "sync"
)

func squareWorker(in <-chan int, out chan<- int, wg *sync.WaitGroup) {
    defer wg.Done()
    for num := range in {
        out <- num * num
    }
}

func main() {
    data := []int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}
    const numWorkers = 3

    inCh := make(chan int)
    outCh := make(chan int)
    var wg sync.WaitGroup

    for i := 0; i < numWorkers; i++ {
        wg.Add(1)
        go squareWorker(inCh, outCh, &wg)
    }

    go func() {
        for _, num := range data {
            inCh <- num
        }
        close(inCh)
    }()

    go func() {
        wg.Wait()
        close(outCh)
    }()

    for result := range outCh {
        fmt.Println(result)
    }
}

在上述代码中，我们创建了3个squareWorker Goroutine来并行计算平方。主Goroutine将数据发送到inCh，squareWorker从inCh接收数据并将计算结果发送到outCh。最后，主Goroutine从outCh接收并打印结果。

扇入（Fan - In）

扇入是指将多个Goroutine的结果合并到一个结果集中。结合上面扇出的例子，我们可以将多个squareWorker的结果合并。

package main

import (
    "fmt"
    "sync"
)

func squareWorker(in <-chan int, out chan<- int, wg *sync.WaitGroup) {
    defer wg.Done()
    for num := range in {
        out <- num * num
    }
}

func fanIn(inputs []<-chan int, out chan<- int, wg *sync.WaitGroup) {
    defer wg.Done()
    var wgInner sync.WaitGroup
    for _, in := range inputs {
        wgInner.Add(1)
        go func(c <-chan int) {
            defer wgInner.Done()
            for val := range c {
                out <- val
            }
        }(in)
    }
    go func() {
        wgInner.Wait()
        close(out)
    }()
}

func main() {
    data := []int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}
    const numWorkers = 3

    var wg sync.WaitGroup
    inputChannels := make([]<-chan int, numWorkers)
    for i := 0; i < numWorkers; i++ {
        inCh := make(chan int)
        outCh := make(chan int)
        inputChannels[i] = outCh
        wg.Add(1)
        go squareWorker(inCh, outCh, &wg)

        go func() {
            for _, num := range data[i::numWorkers] {
                inCh <- num
            }
            close(inCh)
        }()
    }

    finalOut := make(chan int)
    wg.Add(1)
    go fanIn(inputChannels, finalOut, &wg)

    go func() {
        wg.Wait()
        close(finalOut)
    }()

    for result := range finalOut {
        fmt.Println(result)
    }
}

在这个例子中，fanIn函数将多个squareWorker的输出Channel合并到一个finalOut Channel中。这样，我们就实现了扇入的功能，将多个并行任务的结果汇总。

错误处理与并发

在并发编程中，错误处理变得更加复杂。因为多个Goroutine可能同时产生错误，我们需要一种有效的方式来处理这些错误。

一种常见的做法是使用一个专门的Channel来传递错误信息。例如：

package main

import (
    "fmt"
    "sync"
)

func worker(id int, errCh chan<- error) {
    // 模拟可能出错的操作
    if id == 2 {
        errCh <- fmt.Errorf("worker %d encountered an error", id)
    } else {
        errCh <- nil
    }
}

func main() {
    var wg sync.WaitGroup
    errCh := make(chan error)

    for i := 0; i < 3; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            worker(id, errCh)
        }(i)
    }

    go func() {
        wg.Wait()
        close(errCh)
    }()

    for err := range errCh {
        if err != nil {
            fmt.Println(err)
        }
    }
}

在上述代码中，每个worker Goroutine将可能产生的错误发送到errCh Channel。主Goroutine通过遍历errCh来获取并处理这些错误。

性能优化与并发

在使用Goroutine进行并发编程时，性能优化是一个重要的方面。以下是一些常见的性能优化点：

合理设置Goroutine数量

创建过多的Goroutine会增加调度开销，导致性能下降。可以根据任务的类型和系统资源来合理设置Goroutine的数量。例如，对于CPU密集型任务，可以设置Goroutine数量为CPU核心数；对于I/O密集型任务，可以适当增加Goroutine数量以充分利用I/O空闲时间。

减少锁的竞争

Mutex等锁机制虽然能保证数据安全，但过多的锁竞争会降低并发性能。可以通过设计合理的数据结构和算法，尽量减少锁的使用范围和时间。例如，使用无锁数据结构（如sync.Map在某些场景下可以替代map加锁的方式）。

优化Channel使用

合理设置Channel的缓冲区大小可以避免不必要的阻塞和性能开销。对于生产者 - 消费者模型，根据生产和消费的速度来调整缓冲区大小，以达到最佳性能。

总结

Goroutine并发模型是Go语言的一大特色，通过Goroutine、Channel、Mutex、WaitGroup等机制，我们可以方便地实现高效的并发编程。在实际应用中，需要根据具体的业务场景，合理运用这些机制，解决并发通信、同步、错误处理等问题，并进行性能优化。从简单的生产者 - 消费者模型到复杂的扇入扇出模型，Goroutine并发模型都能提供有效的解决方案，帮助开发者充分利用多核处理器的性能，提高程序的运行效率。