使用goroutines与sync.WaitGroup统计质数结果不一致问题求助

问题背景 #

尝试用Go的goroutines统计小于等于整数i的质数数量，比如i=100时正确结果是25。但当i较大（如10000、1000000）时，每次运行结果都不一致且错误，数值波动随i增大而明显。

原实现代码：

package main

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

var wg sync.WaitGroup

func isprime(x int) bool {
    if x == 2 {
        return true
    }
    if x == 1 || x%2 == 0 {
        return false
    }
    var xi = float64(x)
    for i := 3; float64(i) < (math.Pow(xi, 0.5) + 1.0); i += 2 {
        if x%i == 0 {
            return false
        }
    }
    return true
}

func main() {
    fmt.Print("Till what number should I count primes? ")
    var i int
    fmt.Scan(&i)

    r := 0
    pr := &r
    fmt.Println("Counting primes till ", i)
    start := time.Now()
    for x := 0; x <= i; x++ {
        wg.Add(1)
        go func(n int) {
            defer wg.Done()
            if isprime(n) {
                *pr += 1
            }
        }(x)
    }
    wg.Wait()
    elapsed := time.Since(start).Seconds()
    fmt.Println("Counted", r, "primes")
    fmt.Println("took", elapsed, "seconds")
}

运行异常示例：

❯ ./main
Till what number should I count primes? 10000
Counting primes till  10000
Counted 1228 primes
took 0.006776541 seconds
❯ ./main
Till what number should I count primes? 10000
Counting primes till  10000
Counted 1227 primes
took 0.004183875 seconds
❯ ./main
Till what number should I count primes? 1000000
Counting primes till  1000000
Counted 78254 primes
took 0.441985921 seconds
❯ ./main
Till what number should I count primes? 1000000
Counting primes till  1000000
Counted 78327 primes
took 0.430042047 seconds

问题原因 #

核心是竞态条件（Race Condition）：多个goroutine同时对共享变量r（通过指针pr访问）执行*pr +=1操作，而这个操作不是原子的——它会被Go编译器拆成三个步骤：

1. 读取*pr的当前值到寄存器
2. 寄存器中的值加1
3. 将新值写回*pr

当多个goroutine同时执行这三步时，会出现互相覆盖的情况：比如goroutine A读取了值10，goroutine B也读取了值10，两者都加1变成11，然后都写回，最终*pr只增加了1，而不是2，导致计数偏少或波动。

小数值时偶尔正确是因为goroutine数量少，冲突概率低，但数值越大，goroutine越多，冲突概率越高，结果偏差就越明显。

修复方案 #

方案1：使用互斥锁（sync.Mutex） #

通过互斥锁保证同一时间只有一个goroutine能修改共享变量，避免竞态：

修改后的代码：

package main

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

var wg sync.WaitGroup
var mutex sync.Mutex // 新增互斥锁

func isprime(x int) bool {
    if x == 2 {
        return true
    }
    if x == 1 || x%2 == 0 {
        return false
    }
    var xi = float64(x)
    for i := 3; float64(i) < (math.Pow(xi, 0.5) + 1.0); i += 2 {
        if x%i == 0 {
            return false
        }
    }
    return true
}

func main() {
    fmt.Print("Till what number should I count primes? ")
    var i int
    fmt.Scan(&i)

    r := 0
    pr := &r
    fmt.Println("Counting primes till ", i)
    start := time.Now()
    for x := 0; x <= i; x++ {
        wg.Add(1)
        go func(n int) {
            defer wg.Done()
            if isprime(n) {
                mutex.Lock() // 修改前加锁
                *pr += 1
                mutex.Unlock() // 修改后解锁
            }
        }(x)
    }
    wg.Wait()
    elapsed := time.Since(start).Seconds()
    fmt.Println("Counted", r, "primes")
    fmt.Println("took", elapsed, "seconds")
}

方案2：使用原子操作（sync/atomic） #

原子操作由底层硬件支持，能保证+1操作的原子性，性能比互斥锁更优：

修改后的代码：

package main

import (
    "fmt"
    "math"
    "sync"
    "sync/atomic"
    "time"
)

var wg sync.WaitGroup

func isprime(x int) bool {
    if x == 2 {
        return true
    }
    if x == 1 || x%2 == 0 {
        return false
    }
    var xi = float64(x)
    for i := 3; float64(i) < (math.Pow(xi, 0.5) + 1.0); i += 2 {
        if x%i == 0 {
            return false
        }
    }
    return true
}

func main() {
    fmt.Print("Till what number should I count primes? ")
    var i int
    fmt.Scan(&i)

    var r int32 = 0 // 改用int32类型适配原子操作
    fmt.Println("Counting primes till ", i)
    start := time.Now()
    for x := 0; x <= i; x++ {
        wg.Add(1)
        go func(n int) {
            defer wg.Done()
            if isprime(n) {
                atomic.AddInt32(&r, 1) // 原子加1
            }
        }(x)
    }
    wg.Wait()
    elapsed := time.Since(start).Seconds()
    fmt.Println("Counted", r, "primes")
    fmt.Println("took", elapsed, "seconds")
}

方案3：使用通道（Channel）收集结果 #

每个goroutine将质数判断结果发送到通道，主goroutine统一统计，完全避免共享变量：

修改后的代码：

package main

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

var wg sync.WaitGroup

func isprime(x int) bool {
    if x == 2 {
        return true
    }
    if x == 1 || x%2 == 0 {
        return false
    }
    var xi = float64(x)
    for i := 3; float64(i) < (math.Pow(xi, 0.5) + 1.0); i += 2 {
        if x%i == 0 {
            return false
        }
    }
    return true
}

func main() {
    fmt.Print("Till what number should I count primes? ")
    var i int
    fmt.Scan(&i)

    resultChan := make(chan bool, i+1) // 带缓冲的通道避免阻塞
    fmt.Println("Counting primes till ", i)
    start := time.Now()
    for x := 0; x <= i; x++ {
        wg.Add(1)
        go func(n int) {
            defer wg.Done()
            resultChan <- isprime(n)
        }(x)
    }
    // 启动goroutine关闭通道，避免主goroutine阻塞
    go func() {
        wg.Wait()
        close(resultChan)
    }()
    // 统计结果
    r := 0
    for res := range resultChan {
        if res {
            r++
        }
    }
    elapsed := time.Since(start).Seconds()
    fmt.Println("Counted", r, "primes")
    fmt.Println("took", elapsed, "seconds")
}

验证 #

三种方案都能保证结果一致且正确，比如i=10000时正确结果是1229，i=1000000时正确结果是78498，运行后会稳定输出这个数值。

内容的提问来源于stack exchange，提问作者libkush