AI 助理
备案控制台
开发者社区 开发与运维 文章 正文

go| 感受并发编程的乐趣 后篇

2021-12-14 172
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
简介： go| 感受并发编程的乐趣 后篇

这篇 blog 紧接我的上篇 blog - go| 感受并发编程的乐趣 前篇.

学习了 ccmouse - googl工程师 在 慕课网 - 搭建并行处理管道，感受GO语言魅力, 获益匪浅, 也想把这份编程的快乐传递给大家.

强烈推荐一下ccmouse大大的课程, 总能让我生出 Google工程师果然就是不一样 之感, 每次都能从简单的 hello world 开始, 一步步 coding 到教程的主题, 并在过程中给予充分的理由 -- 为什么要一步步变复杂. 同时也会亲身 踩坑 示范, 干货满满.

内容提要:

  • go 实现完整外部排序
  • go 实现集群版(web版)外部排序

另外, ccmouse大大关于语言学习的方法也值得借鉴:

  • 首先, 学习一下语言语法的要点
  • 立刻找一个不那么简单的项目来做, 边做边查文档/stackoverflow

go 实现完整外部排序

先来看看完整外部排序的设计图:

图解: 外部排序

涉及到的功能大部分在上一章都有讲到, 整体流程:

  • 从文件中读取数据: 注意这里使用了 chunk 的设计, 将文件进行分块读取, 而且 chunkSize 的设计很巧妙, 同时支持 全文读取 和 chunk读取
  • 对读取到的 chunk 的数据进行内存排序(快排)
  • 通过递归, 对排序后的 chunk 进行二路归并
  • 将归并后的数据写入到文件中

从协程的角度来看待整个流程:

  • goroutine1 进行 chunk读取, 写入 channel
  • goroutine2 进行 内存排序, 排序后数据写入 channel
  • goroutine3 进行 二路归并, 归并的过程中, 数据不断写入到 channel
  • goroutine4 进行 文件写入, 将 channel 中的数据写入到文件

注意这里:

  • goroutine1-4 可能是多个协程, 可能某一时刻是同一个协程, go 底层会有任务队列(runq)进行协程调度
  • 可以通过数据流的角度来思考这个问题: 数据是怎么在 文件/channel/协程 之间进行流转的.
  • 测试很重要, 示例中就先使用了small 数据进行测试, 检查程序的正确性, 再调整到 large 数据
  • 日志很重要, 可以帮助我们获取到程序的更多信息, 比如 debug/性能调优

关于性能:

  • 并行 最终受限于 cpu 核数, 即 N 核cpu最多同时运行 N 个线程
  • 协程间的抢占会带来性能损耗, 同理还有 进程/线程 的调度
  • 协程+channel的机制方便并发编程扩展, 相对于单机内存操作自然性能要低一些
package main

import (
    "io"
    "encoding/binary"
    "os"
    "bufio"
    "sort"
    "fmt"
    "time"
)

var startTime time.Time

func main() {
    fileIn := "small.in"
    fileOut := "small.out"
    p := createPipeline(fileIn, 512, 4) // 按照cpu核数设置节点数, 减少协程间抢占带来性能损耗
    writeToFile(p, fileOut)
    printFile(fileOut, -1)

    startTime = time.Now() // 添加日志
    fileIn = "large.in"
    fileOut = "large.out"
    p = createPipeline(fileIn, 800000000, 4)
    writeToFile(p, fileOut)
    printFile(fileOut, 100)
}

func createPipeline(filename string, fileSize, chunkCount int) <-chan int {
    chunkSize := fileSize / chunkCount // fileSize/8/chunkCount = int/chunk, 这里简单处理, 设置为可以整除的参数
    sortResults := []<-chan int{}      // 传递给 mergeN() 的已排序切片
    for i := 0; i < chunkCount; i++ {
        file, err := os.Open(filename) // 为什么没有用 defer file.close() ? 因为需要在函数外去关闭掉, 比较麻烦, 这里暂时省略
        if err != nil {
            panic(err)
        }
        file.Seek(int64(i*chunkSize), 0) // 定位到每个 chunk 的起始位置
        s := readerChunk(bufio.NewReader(file), chunkSize)
        sortResults = append(sortResults, memSort(s))
    }
    return mergeN(sortResults...)
}

func writeToFile(ch <-chan int, filename string) {
    file, err := os.Create(filename)
    if err != nil {
        panic(err)
    }
    defer file.Close()
    writer := bufio.NewWriter(file)
    defer writer.Flush() // defer 是 LIFO

    for v := range ch {
        buffer := make([]byte, 8)
        binary.BigEndian.PutUint64(buffer, uint64(v))
        writer.Write(buffer)
    }
}

func printFile(filename string, count int) {
    file, err := os.Open(filename)
    if err != nil {
        panic(err)
    }
    defer file.Close()
    p := readerChunk(file, -1) // -1 的作用体现出来了, 这里就可以读取全部文件
    if count == -1 {
        for v := range p {
            fmt.Println(v)
        }
    } else {
        n := 0
        for v := range p {
            fmt.Println(v)
            n++
            if n >= count {
                break
            }
        }
    }
}

// 递归解决两两归并
func mergeN(ins ...<-chan int) <-chan int {
    if len(ins) == 1 {
        return ins[0]
    }
    m := len(ins) / 2
    // ins[0..m) + ins[m..end)
    return merge(mergeN(ins[:m]...),
        mergeN(ins[m:]...))
}

func merge(in1, in2 <-chan int) <-chan int {
    out := make(chan int, 1024) // 性能优化, 给 channel 添加 buffer, 而不是收一个就发一个
    go func() {
        // 归并的过程要处理某个通道可能没有数据的情况, 代码非常值得一读
        v1, ok1 := <-in1
        v2, ok2 := <-in2
        for ok1 || ok2 {
            if !ok2 || (ok1 && v1 <= v2) {
                out <- v1
                v1, ok1 = <-in1
            } else {
                out <- v2
                v2, ok2 = <-in2
            }
        }
        close(out)
        fmt.Println("merge done: ", time.Now().Sub(startTime))
    }()
    return out
}

// 添加 chunk 来读取文件,
func readerChunk(reader io.Reader, chunkSize int) <-chan int {
    out := make(chan int, 1024) // 性能优化, 给 channel 添加 buffer, 而不是收一个就发一个
    bytesRead := 0
    go func() {
        buffer := make([]byte, 8) // int: 64bit -> 8byte
        for {
            n, err := reader.Read(buffer)
            bytesRead += n
            if n > 0 { // 可能数据不足 8byte
                v := int(binary.BigEndian.Uint64(buffer))
                out <- v
            }
            // 使用 -1 表示不添加 chunk 大小限制
            // 使用是 >=, 读取区间是 [0, chunkSize)
            if err != nil || (chunkSize != -1 && bytesRead >= chunkSize) {
                break
            }
        }
        close(out)
    }()
    return out
}

func memSort(in <-chan int) <-chan int {
    out := make(chan int, 1024) // 性能优化, 给 channel 添加 buffer, 而不是收一个就发一个
    go func() {
        // read into memory
        a := []int{}
        for v := range in {
            a = append(a, v)
        }
        fmt.Println("read into memory: ", time.Now().Sub(startTime))
        // sort
        sort.Ints(a)
        fmt.Println("sort done: ", time.Now().Sub(startTime))
        // output
        for _, v := range a {
            out <- v
        }
        close(out)
    }()
    return out
}

go 实现集群版(web版)外部排序

网络版的设计:

图解: 网络版的修改

网络版只是在完整外排序的版本上, 新增了从网络读写数据, 并相应修改 pipeline 即可

package main

import (
    "net"
    "bufio"
    "encoding/binary"
    "os"
    "strconv"
    "time"
    "fmt"
    "sort"
    "io"
)

var startTime time.Time

func main() {
    startTime = time.Now()

    // 测试 net server
    //netPipeline("small.in", 512, 4) // 按照cpu核数设置节点数, 减少协程间抢占带来性能损耗
    //time.Sleep(time.Hour)

    // 测试 small
    p := netPipeline("small.in", 512, 4)
    writeToFile(p, "small.out")
    printFile("small.out", -1)

    // 测试 large
    //p := netPipeline("small.in", 512, 4)
    //writeToFile(p, "small.out")
    //printFile("small.out", -1)
}

func netPipeline(filename string, fileSize, chunkCount int) <-chan int {
    chunkSize := fileSize / chunkCount // fileSize/8/chunkCount = int/chunk, 这里简单处理, 设置为可以整除的参数
    sortAddr := []string{}
    for i := 0; i < chunkCount; i++ {
        file, err := os.Open(filename) // 为什么没有用 defer file.close() ? 因为需要在函数外去关闭掉, 比较麻烦, 这里暂时省略
        if err != nil {
            panic(err)
        }
        file.Seek(int64(i*chunkSize), 0) // 定位到每个 chunk 的起始位置
        s := readerChunk(bufio.NewReader(file), chunkSize)

        addr := ":" + strconv.Itoa(7000 + i) // 设置不同端口号来设置不同的 server
        netSink(addr, memSort(s)) // 注意 pipeline 的设计思路是建立执行流程, 真正开始执行在 pipeline 创建之后
        sortAddr = append(sortAddr, addr)
    }

    //return nil // 测试 net server

    sortResults := []<-chan int{}
    for _,addr := range sortAddr {
        sortResults = append(sortResults, netSource(addr))
    }
    return mergeN(sortResults...)
}

func writeToFile(ch <-chan int, filename string) {
    file, err := os.Create(filename)
    if err != nil {
        panic(err)
    }
    defer file.Close()
    writer := bufio.NewWriter(file)
    defer writer.Flush() // defer 是 LIFO

    for v := range ch {
        buffer := make([]byte, 8)
        binary.BigEndian.PutUint64(buffer, uint64(v))
        writer.Write(buffer)
    }
}

func printFile(filename string, count int) {
    file, err := os.Open(filename)
    if err != nil {
        panic(err)
    }
    defer file.Close()
    p := readerChunk(file, -1) // -1 的作用体现出来了, 这里就可以读取全部文件
    if count == -1 {
        for v := range p {
            fmt.Println(v)
        }
    } else {
        n := 0
        for v := range p {
            fmt.Println(v)
            n++
            if n >= count {
                break
            }
        }
    }
}

func netSink(addr string, in <-chan int) {
    listener, err := net.Listen("tcp", addr)
    if err != nil {
        panic(err)
    }
    go func() {
        defer listener.Close()
        conn, err := listener.Accept() // 通常 accept() 要放到 for{} 中不断的接收请求, 这里只处理一次就关闭了
        if err != nil {
            panic(err)
        }
        defer conn.Close()
        writer := bufio.NewWriter(conn)
        defer writer.Flush() // 别忘了 flush buffer
        for v := range in {
            buffer := make([]byte, 8)
            binary.BigEndian.PutUint64(buffer, uint64(v))
            writer.Write(buffer)
        }
    }()
}

func netSource(addr string) <-chan int {
    out := make(chan int)
    go func() {
        conn, err := net.Dial("tcp", addr)
        if err != nil {
            panic(err)
        }
        defer conn.Close()
        r := readerChunk(bufio.NewReader(conn), -1)
        for v := range r {
            out <- v
        }
        close(out)
    }()
    return out
}

// 递归解决两两归并
func mergeN(ins ...<-chan int) <-chan int {
    if len(ins) == 1 {
        return ins[0]
    }
    m := len(ins) / 2
    // ins[0..m) + ins[m..end)
    return merge(mergeN(ins[:m]...),
        mergeN(ins[m:]...))
}

func merge(in1, in2 <-chan int) <-chan int {
    out := make(chan int, 1024)
    go func() {
        // 归并的过程要处理某个通道可能没有数据的情况, 代码非常值得一读
        v1, ok1 := <-in1
        v2, ok2 := <-in2
        for ok1 || ok2 {
            if !ok2 || (ok1 && v1 <= v2) {
                out <- v1
                v1, ok1 = <-in1
            } else {
                out <- v2
                v2, ok2 = <-in2
            }
        }
        close(out)
        fmt.Println("merge done: ", time.Now().Sub(startTime))
    }()
    return out
}

// 添加 chunk 来读取文件,
func readerChunk(reader io.Reader, chunkSize int) <-chan int {
    out := make(chan int, 1024) // 性能优化, 给 channel 添加 buffer, 而不是收一个就发一个
    bytesRead := 0
    go func() {
        buffer := make([]byte, 8) // int: 64bit -> 8byte
        for {
            n, err := reader.Read(buffer)
            bytesRead += n
            if n > 0 { // 可能数据不足 8byte
                v := int(binary.BigEndian.Uint64(buffer))
                out <- v
            }
            // 使用 -1 表示不添加 chunk 大小限制
            // 使用是 >=, 读取区间是 [0, chunkSize)
            if err != nil || (chunkSize != -1 && bytesRead >= chunkSize) {
                break
            }
        }
        close(out)
    }()
    return out
}

func memSort(in <-chan int) <-chan int {
    out := make(chan int, 1024)
    go func() {
        // read into memory
        a := []int{}
        for v := range in {
            a = append(a, v)
        }
        fmt.Println("read into memory: ", time.Now().Sub(startTime))
        // sort
        sort.Ints(a)
        fmt.Println("sort done: ", time.Now().Sub(startTime))
        // output
        for _, v := range a {
            out <- v
        }
        close(out)
    }()
    return out
}

写在最后

go 的「强制」在编程方面感觉优点大于缺点:

  • 强制代码风格: 读/写代码都轻松了不少
  • 强制类型检查: 出错时的错误提示非常友好

书写过程中, 基本根据编译器提示, 就可以把大部分 bug 清理掉.

go语言三大特色:

  • 面向接口, 比如示例中的 Reader/Writer, 从而可以轻松添加 buffer 进行性能优化
  • 函数式, go语言中函数式一等公民
  • 并发编程: go + channel

再次推荐一下 go, 给想要写 并发编程 的程序汪, 就如 ccmouse大大的教程所说:

感受并发编程的乐趣

资源推荐:

文章标签：
Go
调度
编译器
关键词：
Go并发编程
Go并发
Go感受并发编程乐趣
陈志林
目录
相关文章
游客sjjs6jgz4gp7m1
|
12天前
|
并行计算 安全 Go
Go语言中的并发编程：掌握goroutines和channels####
本文深入探讨了Go语言中并发编程的核心概念——goroutine和channel。不同于传统的线程模型，Go通过轻量级的goroutine和通信机制channel，实现了高效的并发处理。我们将从基础概念开始，逐步深入到实际应用案例，揭示如何在Go语言中优雅地实现并发控制和数据同步。 ####
游客sjjs6jgz4gp7m1
85 59
叫个什么名字
|
17天前
|
存储 Go 开发者
Go语言中的并发编程与通道（Channel）的深度探索
本文旨在深入探讨Go语言中并发编程的核心概念和实践，特别是通道（Channel）的使用。通过分析Goroutines和Channels的基本工作原理，我们将了解如何在Go语言中高效地实现并行任务处理。本文不仅介绍了基础语法和用法，还深入讨论了高级特性如缓冲通道、选择性接收以及超时控制等，旨在为读者提供一个全面的并发编程视角。
叫个什么名字
82 50
游客762btuqu5wybw666
|
13天前
|
安全 Serverless Go
Go语言中的并发编程：深入理解与实践####
本文旨在为读者提供一个关于Go语言并发编程的全面指南。我们将从并发的基本概念讲起，逐步深入到Go语言特有的goroutine和channel机制，探讨它们如何简化多线程编程的复杂性。通过实例演示和代码分析，本文将揭示Go语言在处理并发任务时的优势，以及如何在实际项目中高效利用这些特性来提升性能和响应速度。无论你是Go语言的初学者还是有一定经验的开发者，本文都将为你提供有价值的见解和实用的技巧。 ####
游客762btuqu5wybw666
29 13
叫个什么名字
|
15天前
|
Go 调度 开发者
Go语言中的并发编程：深入理解goroutines和channels####
本文旨在探讨Go语言中并发编程的核心概念——goroutines和channels。通过分析它们的工作原理、使用场景以及最佳实践，帮助开发者更好地理解和运用这两种强大的工具来构建高效、可扩展的应用程序。文章还将涵盖一些常见的陷阱和解决方案，以确保在实际应用中能够避免潜在的问题。 ####
叫个什么名字
22 3
游客5fdji2pvmf8888
|
17天前
|
安全 Go 数据处理
Go语言中的并发编程：掌握goroutine和channel的艺术####
本文深入探讨了Go语言在并发编程领域的核心概念——goroutine与channel。不同于传统的单线程执行模式，Go通过轻量级的goroutine实现了高效的并发处理，而channel作为goroutines之间通信的桥梁，确保了数据传递的安全性与高效性。文章首先简述了goroutine的基本特性及其创建方法，随后详细解析了channel的类型、操作以及它们如何协同工作以构建健壮的并发应用。此外，还介绍了select语句在多路复用中的应用，以及如何利用WaitGroup等待一组goroutine完成。最后，通过一个实际案例展示了如何在Go中设计并实现一个简单的并发程序，旨在帮助读者理解并掌
游客5fdji2pvmf8888
33 5
游客5fdji2pvmf8888
|
17天前
|
安全 Java Go
Go语言中的并发编程：掌握goroutine与通道的艺术####
本文深入探讨了Go语言中的核心特性——并发编程，通过实例解析goroutine和通道的高效使用技巧，旨在帮助开发者提升多线程程序的性能与可靠性。 ####
游客5fdji2pvmf8888
23 5
游客762btuqu5wybw666
|
18天前
|
Go 开发者
Go语言中的并发编程：掌握goroutines和channels####
本文深入探讨了Go语言中并发编程的核心概念，重点介绍了goroutines和channels的工作原理及其在实际开发中的应用。文章通过实例演示如何有效地利用这些工具来编写高效、可维护的并发程序，旨在帮助读者理解并掌握Go语言在处理并发任务时的强大能力。 ####
游客762btuqu5wybw666
30 6
游客sjjs6jgz4gp7m1
|
16天前
|
算法 安全 程序员
Go语言的并发编程：深入理解与实践####
本文旨在探讨Go语言在并发编程方面的独特优势及其实现机制，通过实例解析关键概念如goroutine和channel，帮助开发者更高效地利用Go进行高性能软件开发。不同于传统的摘要概述，本文将以一个简短的故事开头，引出并发编程的重要性，随后详细阐述Go语言如何简化复杂并发任务的处理，最后通过实际案例展示其强大功能。 --- ###
游客sjjs6jgz4gp7m1
23 2
龙大吉
|
20天前
|
存储 安全 Go
Go 语言以其高效的并发编程能力著称，主要依赖于 goroutines 和 channels 两大核心机制
Go 语言以其高效的并发编程能力著称，主要依赖于 goroutines 和 channels 两大核心机制。本文介绍了这两者的概念、用法及如何结合使用，实现任务的高效并发执行与数据的安全传递，强调了并发编程中的注意事项，旨在帮助开发者更好地掌握 Go 语言的并发编程技巧。
龙大吉
29 2
众所周知
|
21天前
|
存储 负载均衡 监控
如何利用Go语言的高效性、并发支持、简洁性和跨平台性等优势，通过合理设计架构、实现负载均衡、构建容错机制、建立监控体系、优化数据存储及实施服务治理等步骤，打造稳定可靠的服务架构。
在数字化时代，构建高可靠性服务架构至关重要。本文探讨了如何利用Go语言的高效性、并发支持、简洁性和跨平台性等优势，通过合理设计架构、实现负载均衡、构建容错机制、建立监控体系、优化数据存储及实施服务治理等步骤，打造稳定可靠的服务架构。
众所周知
29 1

热门文章

最新文章

  • 1
    【chatgpt】openai api 接口调用（go语言版）
  • 2
    Go——小白学习之接口
  • 3
    Go struct tag能否设置默认值？
  • 4
    Go语言之反射
  • 5
    Go语言之函数详解
  • 6
    GO 之MAP的使用
  • 7
    GO语言练习：组合的用法
  • 8
    Go语言之标志符可见性
  • 9
    掌握Go语言：Go语言中的字典魔法，高效数据检索与应用实例解析（18）
  • 10
    Go如何自动解压缩包？如何读取docx/doc文件内容？
  • 1
    【GO基础】GO基础语法一
    36
  • 2
    【GO基础】2. IDEA配置Go语言开发环境
    647
  • 3
    【GO基础】1. Go语言环境搭建
    64
  • 4
    lucky - go 语言实现的快速开发平台
    58
  • 5
    Go 语言切片如何扩容？（全面解析原理和过程）
    126
  • 6
    使用Go语言构建高性能的Web服务器：协程与Channel的深度解析
    249
  • 7
    构建高可用的分布式数据库集群：使用Go语言与Raft共识算法
    104
  • 8
    基于Go语言的微服务架构实践
    171
  • 9
    使用Go语言进行并发编程
    58
  • 10
    深入理解Go的垃圾回收机制
    117

    • 相关课程

    更多
  • Go语言核心编程 - 数据结构和算法
  • Go语言核心编程 - 基础语法、数组、切片、Map
  • Go语言核心编程 - 面向对象、文件、单元测试、反射、TCP编程
  • Go语言完全自学手册图文教程
  • PHP进阶教程 - 由浅入深掌握面向对象开发 - 第二阶段
  • 「大师课」搞定 Java 开发基础

    • 相关电子书

    更多
  • Go语言路上踩过的坑
  • gohbase ：HBase go客户端
  • Go构建日请求千亿级微服务实践

    • 相关实验场景

    更多
  • 每个IT人都想学的“Web应用上云经典架构”实战
  • 语言入门-1：环境构建
  • 高性能特性体验：ePQ 的详解与实战
  • 1分钟部署经典小游戏
    • 下一篇
    阿里云对象存储OSS收费标准：存储、流量和请求等多个计费项详解