当前位置：首页 > 文章列表 > Golang > Go教程 > 掌握 Go 并发：高性能应用程序的基本模式

掌握 Go 并发：高性能应用程序的基本模式

来源：dev.to 2024-12-20 20:10:01 0浏览收藏

你在学习Golang相关的知识吗？本文《掌握 Go 并发：高性能应用程序的基本模式》，主要介绍的内容就涉及到，如果你想提升自己的开发能力，就不要错过这篇文章，大家要知道编程理论基础和实战操作都是不可或缺的哦！

当涉及到用 go 构建高效且可扩展的应用程序时，掌握并发模式至关重要。 go凭借其轻量级的goroutine和强大的通道，为并发编程提供了理想的环境。在这里，我们将深入研究一些最有效的并发模式，包括 goroutine 池、工作队列和扇出/扇入模式，以及最佳实践和要避免的常见陷阱。

goroutine 池

go 中管理并发的最有效方法之一是使用 goroutine 池。 goroutine 池控制在任何给定时间主动执行的 goroutine 数量，这有助于节省内存和 cpu 时间等系统资源。当您需要同时处理大量任务而又不会压垮系统时，这种方法特别有用。

要实现 goroutine 池，首先要创建固定数量的 goroutine 来形成池。然后，这些 goroutine 会被重用来执行任务，从而减少与不断创建和销毁 goroutine 相关的开销。这是一个如何实现 goroutine 池的简单示例：

package main

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

type job func()

func worker(id int, jobs <-chan job, wg *sync.waitgroup) {
    defer wg.done()
    for job := range jobs {
        fmt.printf("worker %d starting job\n", id)
        job()
        fmt.printf("worker %d finished job\n", id)
    }
}

func main() {
    jobs := make(chan job, 100)
    var wg sync.waitgroup

    // start 5 workers.
    for i := 1; i <= 5; i++ {
        wg.add(1)
        go worker(i, jobs, &wg)
    }

    // enqueue 20 jobs.
    for j := 1; j <= 20; j++ {
        job := func() {
            time.sleep(2 * time.second) // simulate time-consuming task
            fmt.println("job completed")
        }
        jobs <- job
    }

    close(jobs) // close the channel to indicate that no more jobs will be added.
    wg.wait()  // wait for all workers to finish.
    fmt.println("all jobs have been processed")
}

正确调整泳池大小

确定 goroutine 池的最佳大小至关重要。 goroutine 太少可能无法充分利用 cpu，而太多则可能导致争用和高开销。您需要根据工作负载和系统容量平衡池大小。使用 pprof 等工具监控性能可以帮助您根据需要调整池大小。

工作队列设计和管理

工作队列本质上是一个管理池中 goroutine 之间任务分配的通道。对该队列的有效管理可确保任务均匀分配，防止某些 goroutine 过载而其他 goroutine 空闲。

以下是设计工作队列的方法：

package main

import (
    "fmt"
    "sync"
)

type worker struct {
    id       int
    jobqueue chan job
    wg       *sync.waitgroup
}

func newworker(id int, jobqueue chan job, wg *sync.waitgroup) *worker {
    return &worker{id: id, jobqueue: jobqueue, wg: wg}
}

func (w *worker) start() {
    defer w.wg.done()
    for job := range w.jobqueue {
        fmt.printf("worker %d starting job\n", w.id)
        job()
        fmt.printf("worker %d finished job\n", w.id)
    }
}

func main() {
    jobqueue := make(chan job, 100)
    var wg sync.waitgroup

    // start 5 workers.
    for i := 1; i <= 5; i++ {
        wg.add(1)
        worker := newworker(i, jobqueue, &wg)
        go worker.start()
    }

    // enqueue 20 jobs.
    for j := 1; j <= 20; j++ {
        job := func() {
            fmt.println("job completed")
        }
        jobqueue <- job
    }

    close(jobqueue) // close the channel to indicate that no more jobs will be added.
    wg.wait()       // wait for all workers to finish.
    fmt.println("all jobs have been processed")
}

扇出/扇入模式

扇出/扇入模式是一种用于并行化和协调并发任务的强大技术。该模式由两个主要阶段组成：扇出和扇入。

扇出

在扇出阶段，单个任务被分成多个可以并发执行的较小的子任务。每个子任务都分配给一个单独的 goroutine，允许并行处理。

扇入

在扇入阶段，所有并发执行的子任务的结果或输出被收集并组合成一个结果。此阶段等待所有子任务完成并汇总其结果。

下面是如何实现扇出/扇入模式以同时将数字加倍的示例：

package main

import (
    "fmt"
    "sync"
)

func doublenumber(num int) int {
    return num * 2
}

func main() {
    numbers := []int{1, 2, 3, 4, 5}
    jobs := make(chan int)
    results := make(chan int)

    var wg sync.waitgroup

    // start 5 worker goroutines.
    for i := 0; i < 5; i++ {
        wg.add(1)
        go func() {
            defer wg.done()
            for num := range jobs {
                result := doublenumber(num)
                results <- result
            }
        }()
    }

    // send jobs to the jobs channel.
    go func() {
        for _, num := range numbers {
            jobs <- num
        }
        close(jobs)
    }()

    // collect results from the results channel.
    go func() {
        wg.wait()
        close(results)
    }()

    // print the results.
    for result := range results {
        fmt.println(result)
    }
}

同步原语

waitgroup、mutex 和 channels 等同步原语对于协调 goroutines 和确保并发程序正确运行至关重要。

等待组

waitgroup 用于等待一组 goroutine 完成。使用方法如下：

package main

import (
    "fmt"
    "sync"
)

func main() {
    var wg sync.waitgroup
    for i := 0; i < 5; i++ {
        wg.add(1)
        go func(id int) {
            defer wg.done()
            fmt.printf("worker %d is working\n", id)
            // simulate work
            time.sleep(2 * time.second)
            fmt.printf("worker %d finished\n", id)
        }(i)
    }
    wg.wait()
    fmt.println("all workers have finished")
}

互斥体

互斥体用于保护共享资源免遭并发访问。这是一个例子：

package main

import (
    "fmt"
    "sync"
)

type counter struct {
    mu    sync.mutex
    count int
}

func (c *counter) increment() {
    c.mu.lock()
    c.count++
    c.mu.unlock()
}

func (c *counter) getcount() int {
    c.mu.lock()
    defer c.mu.unlock()
    return c.count
}

func main() {
    counter := &counter{}
    var wg sync.waitgroup

    for i := 0; i < 100; i++ {
        wg.add(1)
        go func() {
            defer wg.done()
            counter.increment()
        }()
    }

    wg.wait()
    fmt.println("final count:", counter.getcount())
}

处理正常关闭

优雅的关闭在并发系统中至关重要，以确保所有正在进行的任务在程序退出之前完成。以下是如何使用退出信号处理正常关闭：

package main

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

func worker(id int, quit <-chan bool, wg *sync.waitgroup) {
    defer wg.done()
    for {
        select {
        case <-quit:
            fmt.printf("worker %d received quit signal\n", id)
            return
        default:
            fmt.printf("worker %d is working\n", id)
            time.sleep(2 * time.second)
        }
    }
}

func main() {
    quit := make(chan bool)
    var wg sync.waitgroup

    for i := 1; i <= 5; i++ {
        wg.add(1)
        go worker(i, quit, &wg)
    }

    time.sleep(10 * time.second)
    close(quit) // send quit signal
    wg.wait()   // wait for all workers to finish
    fmt.println("all workers have finished")
}

基准测试和优化并发代码

基准测试对于了解并发代码的性能至关重要。 go 提供了一个内置的测试包，其中包括基准测试工具。

以下是如何对简单并发函数进行基准测试的示例：

package main

import (
    "testing"
    "time"
)

func concurrentwork() {
    var wg sync.waitgroup
    for i := 0; i < 100; i++ {
        wg.add(1)
        go func() {
            defer wg.done()
            time.sleep(2 * time.second)
        }()
    }
    wg.wait()
}

func benchmarkconcurrentwork(b *testing.b) {
    for i := 0; i < b.n; i++ {
        concurrentwork()
    }
}

要运行基准测试，您可以使用带 -bench 标志的 go test 命令：

go test -bench=. -benchmem -benchtime=10s

错误处理策略

由于 goroutine 的异步特性，并发程序中的错误处理可能具有挑战性。以下是一些有效处理错误的策略：

使用渠道

您可以使用通道将错误从 goroutine 传播到主 goroutine。

package main

import (
    "fmt"
    "sync"
)

func worker(id int, jobqueue <-chan job, errorqueue chan<- error, wg *sync.waitgroup) {
    defer wg.done()
    for job := range jobqueue {
        if err := job(); err != nil {
            errorqueue <- err
        }
    }
}

func main() {
    jobqueue := make(chan job, 100)
    errorqueue := make(chan error, 100)
    var wg sync.waitgroup

    for i := 1; i <= 5; i++ {
        wg.add(1)
        go worker(i, jobqueue, errorqueue, &wg)
    }

    // enqueue jobs.
    for j := 1; j <= 20; j++ {
        job := func() error {
            // simulate an error.
            if j == 10 {
                return fmt.errorf("job %d failed", j)
            }
            return nil
        }
        jobqueue <- job
    }

    close(jobqueue) // close the job queue.

    go func() {
        wg.wait()
        close(errorqueue) // close the error queue.
    }()

    for err := range errorqueue {
        fmt.println("error:", err)
    }
}

使用上下文

context 包提供了一种取消操作并在 goroutine 之间传播错误的方法。

package main

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

func worker(ctx context.Context, id int, jobQueue <-chan Job, wg *sync.WaitGroup) {
    defer wg.Done()
    for {
        select {
        case <-ctx.Done():
            fmt.Printf("Worker %d received cancel signal\n", id)
            return
        case job, ok := <-jobQueue:
            if !ok {
                return
            }
            if err := job(); err != nil {
                fmt.Printf("Worker %d encountered error: %v\n", id, err)
            }
        }
    }
}

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    jobQueue := make(chan Job, 100)
    var wg sync.WaitGroup

    for i := 1; i <= 5; i++ {
        wg.Add(1)
        go worker(ctx, i, jobQueue, &wg)
    }

    // Enqueue jobs.
    for j := 1; j <= 20; j++ {
        job := func() error {
            // Simulate an error.
            if j == 10 {
                return fmt.Errorf("job %d failed", j)
            }
            return nil
        }
        jobQueue <- job
    }

    time.Sleep(10 * time.Second)
    cancel() // Cancel the context.
    close(jobQueue) // Close the job queue.
    wg.Wait()       // Wait for all workers to finish.
}

总之，掌握 go 中的并发模式对于构建健壮、可扩展且高效的应用程序至关重要。通过理解和实现 goroutine 池、工作队列、扇出/扇入模式并使用适当的同步原语，您可以显着增强并发系统的性能和可靠性。始终记住优雅地处理错误并对代码进行基准测试以确保最佳性能。通过这些策略，您可以充分利用 go 并发功能的潜力来构建高性能应用程序。