递归+协程并发+多核

---

如何选定合适的协程数？

N核服务器，通过执行业务的单协程分析出本地计算时间为x，等待时间为y，则工作协程数（协程池协程数）设置为 N*(x+y)/x，能让CPU的利用率最大化。

代码示例：

指定N个协程版：（从头到尾都是N个协程不多也不少）

func TestGoN(nums []int) {
    // 指定协程数
    goroutine_nums := 50
    var c []chan bool
    for i := 0; i < goroutine_nums; i++ {
        ch := make(chan bool)
        c = append(c, ch)
    }

    for i := 0; i < goroutine_nums; i++ {
        go TestgoN(nums, len(nums)/goroutine_nums*i, len(nums)/goroutine_nums*(i+1)-1, c[i])
    }

    for i := 0; i < goroutine_nums; i++ {
        <-c[i]
    }
}

无脑加协程版：（在数据量在1000以下有优化改为局部单协程）

func Testgo(nums []int, left int, right int, ch chan bool) {
    if right <= left {
        ch <- true
        return
    }

    // 1kw级别 递归深度太深，为了比较两个数字的大小
    // 还单独创建一个协程计算不划算。不优化差不多能比正常算法慢1600倍
    if right-left < 1000 {
        test(nums, left, (left+right)/2)
        test(nums, (left+right)/2+1, right)
        ch <- true
        return
    }

    chnext := make(chan bool)

    go Testgo(nums, left, (left+right)/2, chnext)
    go Testgo(nums, (left+right)/2+1, right, chnext)

    <-chnext
    <-chnext
    ch <- true
    return
}

普通方法：

// 无goroutine的普通方法
func test(nums []int, left int, right int) {
    if right <= left {
        return
    }

    test(nums, left, (left+right)/2)
    test(nums, (left+right)/2+1, right)

    return
}

协程数量对效果的影响：

协程数量太多协程切换开销导致效率反倒不如单协程。全协程模式在递归过深的时候有优化所以没有断崖式下跌。

协程数量太少无法充分利用多核无法充分体现多核优势。

四核八线程，横坐标是协程数量，1百万数据量下以常规方法为基的速度比（一百次测试取平均可能存在误差主要给大家看个趋势）

数据量大小对效果的影响：

可见在百万规模以下进行并发效果不见得好。

四核八线程，N为50，横坐标为数据量大小

完整版代码

package main

import (
    "fmt"
    "math/rand"
    "time"
)

func TestGoN(nums []int) {
    // 指定协程数
    goroutine_nums := 4
    var c []chan bool
    for i := 0; i < goroutine_nums; i++ {
        ch := make(chan bool)
        c = append(c, ch)
    }

    for i := 0; i < goroutine_nums; i++ {
        go TestgoN(nums, len(nums)/goroutine_nums*i, len(nums)/goroutine_nums*(i+1)-1, c[i])
    }

    for i := 0; i < goroutine_nums; i++ {
        <-c[i]
    }
}
func TestgoN(nums []int, left int, right int, ch chan bool) {
    test(nums, left, right)
    ch <- true
    return
}

// 纯goroutine方法
func TestGo(nums []int) {
    ch := make(chan bool)

    go Testgo(nums, 0, len(nums)-1, ch)
    // ch是无缓冲channel所以必须用go要不然在return前会被ch<-true阻塞
    <-ch
}
func Testgo(nums []int, left int, right int, ch chan bool) {
    if right <= left {
        ch <- true
        return
    }

    // 1kw级别 递归深度太深，为了比较两个数字的大小还单独创建一个协程计算不划算
    // 不加非goroutine的处理，差不多能比正常算法慢1600倍
    if right-left < 1000 {
        test(nums, left, (left+right)/2)
        test(nums, (left+right)/2+1, right)
        ch <- true
        return
    }

    chnext := make(chan bool)
    // 只需要一个channel，在go里面两次 ch<-true 给外体的两个 <-ch

    go Testgo(nums, left, (left+right)/2, chnext)
    go Testgo(nums, (left+right)/2+1, right, chnext)

    <-chnext
    <-chnext
    ch <- true
    return
}

// 无goroutine的普通方法
func test(nums []int, left int, right int) {
    if right <= left {
        return
    }

    test(nums, left, (left+right)/2)
    test(nums, (left+right)/2+1, right)

    return
}
func Test(nums []int) {
    test(nums, 0, len(nums)-1)
}

type anyfunc func([]int)

func CountTime(f anyfunc) float64 {
    // 利用函数指针传入的方式计算耗时并返回耗时us
    var sum int64
    sum = 0
    times := 1000

    for i := 0; i < times; i++ {
        // 每次生成amount大小的随机数组
        amount := 1000
        t := make([]int, amount)

        for i := 0; i < amount; i++ {
            t[i] = rand.Intn(100)
        }
        start_time := time.Now().UnixNano()
        f(t)
        end_time := time.Now().UnixNano() - start_time
        sum += end_time
    }
    return float64(sum) / float64(times) / 1e6
}
func main() {
    time1 := CountTime(TestGoN)
    fmt.Println("go4 count time ", time1)

    time2 := CountTime(TestGo)
    fmt.Println("go count time ", time2)

    time3 := CountTime(Test)
    fmt.Println("normal count time ", time3)

    fmt.Println("propotion:", time3/time1, time3/time2)
}