社区首页 >专栏 >Go语言的多线程实现与应用案例

Go语言的多线程实现与应用案例

原创

GeekLiHua

发布于 2024-08-30 14:52:07

2940

文章被收录于专栏：gogo golang学习

Go语言的多线程实现与应用案例

1. Goroutine

Go语言中的Goroutine是一种轻量级的线程，它比传统的操作系统线程更轻量级，更易于管理。Goroutine的创建和调度由Go语言的运行时系统自动管理，开发者只需关注业务逻辑的实现。

创建Goroutine

在Go语言中，创建一个Goroutine非常简单，只需要在函数前加上go关键字即可。例如：

package main

import (
    "fmt"
    "time"
)

func main() {
    // 创建一个匿名函数并将其作为Goroutine运行
    go func() {
        // 打印消息，表示Goroutine正在执行
        fmt.Println("Hello from Goroutine")
    }()
    
    // 主函数暂停1秒，以便观察Goroutine的执行效果
    time.Sleep(1 * time.Second)
    
    // 打印消息，表示主函数继续执行
    fmt.Println("Main function continues")
}

在这个例子中，我创建了一个匿名函数，并将其作为Goroutine运行。Goroutine会并行执行，而主函数会继续执行。

2. Channel

Channel是Go语言中用于在Goroutine之间进行通信的机制。它允许Goroutine之间传递数据，并且可以同步执行。Channel可以是缓冲的或非缓冲的。

创建Channel

创建一个Channel非常简单，可以使用make函数。例如：

package main

import "fmt"

func main() {
    // 创建一个int类型的Channel
    ch := make(chan int)
    
    // 向Channel发送数据
    ch <- 42
    
    // 从Channel接收数据
    v := <-ch
    
    // 打印接收到的数据
    fmt.Println(v)
}

发送和接收数据

发送数据到Channel使用<-操作符，接收数据使用<-chan或chan<-类型。

package main

import "fmt"

func main() {
    // 创建一个int类型的Channel
    ch := make(chan int)
    
    // 向Channel发送数据
    ch <- 42
    
    // 从Channel接收数据
    v := <-ch
    
    // 打印接收到的数据
    fmt.Println("Received from channel:", v)
}

3. 同步与等待

在多线程编程中，同步和等待是非常重要的。Go语言提供了多种机制来实现同步和等待。

WaitGroup

sync.WaitGroup是一个计数器，用于等待多个Goroutine完成。每个Goroutine开始时调用Add(1)，结束时调用Done()。

package main

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

func worker(id int, wg *sync.WaitGroup) {
    // 延迟调用Done，确保在函数结束时减少WaitGroup的计数
    defer wg.Done()
    
    // 打印工作开始的消息
    fmt.Printf("Worker %d starting\n", id)
    
    // 模拟工作耗时
    time.Sleep(time.Second)
    
    // 打印工作完成的消息
    fmt.Printf("Worker %d done\n", id)
}

func main() {
    // 创建一个WaitGroup实例
    var wg sync.WaitGroup
    
    // 循环创建并启动5个Goroutine
    for i := 1; i <= 5; i++ {
        wg.Add(1) // 增加WaitGroup的计数
        go worker(i, &wg)
    }
    
    // 等待所有Goroutine完成
    wg.Wait()
    
    // 打印所有工作完成的消息
    fmt.Println("All workers completed")
}

在这个例子中，我创建了5个Goroutine，每个Goroutine执行worker函数，并在完成后调用Done()。主函数通过调用Wait()等待所有Goroutine完成。

4. 应用案例

案例1：并发下载文件

假设我需要并发下载多个文件，可以使用Goroutine和Channel来实现。

package main

import (
    "fmt"
    "io/ioutil"
    "net/http"
    "sync"
)

// download函数用于并发下载文件
func download(url string, wg *sync.WaitGroup, ch chan string) {
    defer wg.Done() // 确保在函数结束时减少WaitGroup的计数
    
    resp, err := http.Get(url) // 发起HTTP GET请求
    if err != nil {
        ch <- err.Error() // 将错误信息发送到Channel
        return
    }
    defer resp.Body.Close() // 确保响应体关闭
    
    body, err := ioutil.ReadAll(resp.Body) // 读取响应体内容
    if err != nil {
        ch <- err.Error() // 将错误信息发送到Channel
        return
    }
    
    ch <- string(body) // 将文件内容发送到Channel
}

func main() {
    urls := []string{
        "http://example.com/file1",
        "http://example.com/file2",
        "http://example.com/file3",
    }
    ch := make(chan string) // 创建一个字符串类型的Channel
    wg := &sync.WaitGroup{} // 创建一个WaitGroup实例
    
    // 循环并发下载文件
    for _, url := range urls {
        wg.Add(1) // 增加WaitGroup的计数
        go download(url, wg, ch)
    }
    
    go func() {
        wg.Wait() // 等待所有Goroutine完成
        close(ch) // 关闭Channel
    }()
    
    // 接收并打印下载结果
    for result := range ch {
        fmt.Println(result)
    }
}

在这个例子中，我并发下载了三个文件，并通过Channel接收下载结果。

案例2：并发计算数据集的平方根

在这个例子中，我将：

创建一个包含大量数据的切片。
使用Goroutine并发计算每个数据元素的平方根。
使用Channel收集结果，并在主函数中打印出来。

package main

import (
    "fmt"
    "math"
    "sync"
)

// process函数用于并发计算数据的平方根
func process(data []float64, ch chan<- float64) {
    for _, v := range data {
        // 计算平方根并发送到Channel
        ch <- math.Sqrt(v)
    }
    close(ch) // 处理完所有数据后关闭Channel
}

func main() {
    // 创建一个包含大量数据的切片
    data := make([]float64, 10000)
    for i := 0; i < 10000; i++ {
        data[i] = float64(i)
    }

    // 创建一个float64类型的Channel用于收集结果
    results := make(chan float64)

    // 创建一个WaitGroup实例用于同步
    var wg sync.WaitGroup

    // 定义每个Goroutine处理的数据量
    chunkSize := 1000

    // 循环并发处理数据
    for i := 0; i < 10000; i += chunkSize {
        end := i + chunkSize
        if end > 10000 {
            end = 10000
        }

        // 创建一个新的Channel用于当前Goroutine的结果
        chunkResults := make(chan float64)

        // 增加WaitGroup的计数
        wg.Add(1)

        // 启动Goroutine处理数据
        go func(start, end int) {
            defer wg.Done() // 确保在函数结束时减少WaitGroup的计数

            // 调用process函数处理数据，并发送结果到chunkResults
            process(data[start:end], chunkResults)

            // 将chunkResults的结果发送到results
            for result := range chunkResults {
                results <- result
            }
        }(i, end)
    }

    // 等待所有Goroutine完成
    go func() {
        wg.Wait()
        close(results) // 所有Goroutine完成后关闭results Channel
    }()

    // 接收并打印结果
    for result := range results {
        fmt.Printf("Result: %.2f\n", result)
    }
}