使用 Go 中的并发提升运行时效率

一分耕耘，一分收获！既然都打开这篇《使用 Go 中的并发提升运行时效率》，就坚持看下去，学下去吧！本文主要会给大家讲到等等知识点，如果大家对本文有好的建议或者看到有不足之处，非常欢迎大家积极提出！在后续文章我会继续更新Golang相关的内容，希望对大家都有所帮助！

我正在用简单的问题示例尝试go中的并发，即第n个素数回文数，例如：第1到第9个回文序列是2,3,5,7,11,101,131,151。我被卡住了并且有不知道该怎么办。

我当前的代码是这样的：

n := 9999999

count := 0

primePalindrome := 0

for i := 0; count < n; i++ {
    go PrimeNumber(i)
    go PalindromeNumber(i)
    
    if PrimeNumber(i) && PalindromeNumber(i) {
        primePalindrome = i
        count++
    }
}

我如何知道函数 primenumber(i) 和 palindromenumber(i) 已经由 go 例程执行并完成，以便我可以给出 if condition 来获取数字？

正确答案

这是我的解决方案：https://go.dev/play/p/KShMctUK9yg

问题是，如果我想使用 go 并发来找到运行速度更快的第 9999999 个回文，如何在 primenumber() 和 palindromenumber() 函数中应用并发？

与 torek 类似，我通过创建多个工作线程来添加并发性，每个工作线程可以检查一个数字，从而并行检查多个数字。

从算法上来说，程序是这样工作的。一个 goroutine 生成可能的素数回文。多个 goroutine 池都会检查候选者。主 goroutine 收集结果。当我们至少有足够的结果来提供nth素数回文时，我们对列表进行排序，然后返回答案。

仔细观察通道和等待组在 goroutine 之间进行通信的使用：

• 多个工作协程。他们的工作是运行回文，然后对候选数进行素数检查。他们收听 candidates 频道；当通道关闭时，它们结束，向 wgsync.waitgroup 传达它们已完成的信息。

• 单个候选生成器。该 goroutine 通过发送到 candidates 通道将每个候选者发送给其中一个工作人员。如果发现done通道关闭，则结束。

• 主协程还充当结果收集器。它从 resc（结果）通道读取并将结果添加到列表中。当列表至少达到所需长度时，它关闭 done，指示生成器停止生成候选。

done 通道可能看起来多余，但它很重要，因为 main 结果收集器知道我们何时完成候选生成，但不是发送到 candidates 的通道。如果我们在 main 中关闭 candidates ，那么生成器很可能会尝试写入它，这会使程序崩溃。生成器是编写候选者的人；它是唯一“知道”不会再编写更多候选的 goroutine。

请注意，此实现至少生成 n 个素数回文。由于它并行生成素数回文，因此不能保证它们是按顺序排列的。在最坏的情况下，我们可能会生成最多质数回文 n+m，其中 m 是工人数量减一。

这里还有很大的改进空间。我非常确定 generator 和 collector 角色可以组合在候选通道和结果通道上的一个 select 循环中。当我在 windows 计算机上运行该程序时，如果 n 与 9999999 一样大，该程序似乎也会遇到非常困难的情况 - 看看您的结果是否有所不同。

编辑：性能增强

如果您希望提高性能，这里有我昨晚发现并注意到的一些事情。

• 无需检查偶数。 for i := 开始; ; i += 1 + (i % 2) 跳到下一个奇数，然后每隔一次加 2 以保持奇数。

• all palindromes of even numbered decimal length are divisble by 11。因此可以跳过整组偶数长度的数字。我通过将 math.pow10(len(str)) 添加到偶数十进制表示来添加另一个数字来实现此目的。这就是导致程序长时间停止输出数字的原因 - 每个偶数组数字都不能产生素数回文。

• 如果数字的十进制表示法以偶数开头，则它不能是素数回文，除非它只有一位数长。 5 也是如此。在下面的代码中，我将 math.pow10(len(str)-1) 添加到要移动到下一个奇数编号序列的数字中。如果数字以 5 开头，我会将其加倍以移至下一个奇数编号序列。

这些技巧使代码的性能提高了很多，但归根结底它仍然是一种蛮力，我仍然没有接近 9999999。

// send candidates
  go func() {
    // send candidates out
    for i := start; ; i += 1 + (i % 2) {
      str := strconv.formatint(i, 10)
      if len(str) % 2 == 0 && i != 11 {
        newi := int64(math.pow10(len(str)))+1
        log.printf("%d => %d", i, newi)
        i = newi
        continue
      }
      if first := (str[0]-'0'); first % 2 == 0 || first == 5 {
        if i < 10 {
          continue
        }
        nexti :=  int64(math.pow10(len(str)-1))
        if first == 5 {
          nexti *= 2
        }
        newi := i + nexti
        log.printf("%d -> %d", i, newi)
        i = newi-2
        continue
      }
      select {
      case _, ok := <-done:
        if !ok {
          close(candidates)
          return
        }

      case candidates <- i:
        continue
      }
    }
  }()

这里有多个问题需要解决：

• 我们想要剥离“是素数”和“是回文”测试
• 我们想要按顺序排列通过测试的数字
• 当然，我们必须用我们的编程语言（在本例中为 go）来表达问题的“衍生”和“等待结果”部分。

（除此之外，我们可能想优化我们的素性测试，也许使用 Sieve of Eratothenese 算法或类似的算法，这也可能涉及并行性。）

这里的中间问题可能是最难的一个。不过，有一种相当明显的方法可以做到这一点：我们可以观察到，如果我们为每个测试的数字分配一个升序数字，则返回的答案 (n适合/不适合），即使它们以错误的顺序返回，也很容易重新调整顺序。

由于您的整体循环增加了 1（这是一种错误¹），因此测试的数字本身就适合此目的。所以我们应该创建一个 go 通道，其类型是一对结果：这是我测试的数字，这是我的答案：

type result struct {
    tested int  // the number tested
    passed bool // pass/fail result
}
testc := make(chan int)
resultc := make(chan result)

接下来，我们将使用典型的 "pool of workers"。然后我们运行要测试的事物循环。这是您现有的循环：

for i := 0; count < n; i++ {
    go primenumber(i)
    go palindromenumber(i)
    
    if primenumber(i) && palindromenumber(i) {
        primepalindrome = i
        count++
    }
}

我们将其重组为：

count := 0
busy := 0
results := []result{}
for totest := 0;; totest += 2 {
    // if all workers are busy, wait for one result
    if busy >= numworkers {
        result := <-resultc // get one result
        busy--
        results := addresult(results, result)
        if result.passed {
            count++ // passed: increment count
            if count >= n {
                break
            }
        }
    }
    // still working, so test this number
    testc <- totest
    busy++
}
close(testc) // tell workers to stop working
// collect remaining results
for result := range resultc {
    results := addresult(results, result)
}

（“繁忙”测试有点笨拙；您可以使用选择来发送或接收，无论您先做什么，但是如果您这样做，下面概述的优化会变得更加复杂。）

这确实意味着我们的标准工作池模式需要关闭结果通道 resultc，这意味着当我们生成 numworkers 工作人员时，我们将添加一个 sync.waitgroup：

var wg sync.WaitGroup
wg.Add(numWorkers)
for i := 0; i < numWorkers; i++ {
    go worker(&wg, testC, resultC)
}
go func() {
    wg.Wait()
    close(resultC)
}()

这使得我们的 for result := range resultc 循环工作；当我们关闭 testc 时，所有工作人员都会停止（并通过 defer 返回并调用 wg.done()，此处未显示），因此一旦最后一个工作人员退出，resultc 就会关闭。

现在我们还有一个问题，那就是：结果以半随机顺序返回。这就是为什么我们有部分结果。 addresult函数需要扩展切片并将结果插入到适当的位置，使用经过测试的值。当主循环到达 break 语句时，totest 中的数字至少是第n个回文素数，但也可能大于第n个。因此，我们需要收集剩余的结果并向后查看某些较早的数字是否实际上是第 n 个。

此时需要进行许多优化：特别是，如果我们通过 k 测试了数字，并且都知道它们已通过或失败，并且 k + numworkers < n，我们不再需要任何这些结果（无论它们通过还是失败），因此我们可以缩小结果切片。或者，如果我们有兴趣构建一个回文素数表，我们可以这样做，或者我们可以选择的任何其他方法。以上并不意味着是最佳解决方案，只是一个解决方案。

再次注意，我们“过冲”：无论 numworkers 是什么，我们都可以测试我们根本不需要测试的 numworkers-1 值。如果每个工作人员正在处理一个高于现在已知的至少第 n 个值的数字。

¹我们可以通过从预加载 2 或 1 和 2 的答案开始（如果您是 choose to consider 1 prime，另请参阅 http://ncatlab.org/nlab/show/too+simple+to+be+simple）来将问题减半。然后我们从 3 向上运行循环 2每次，这样我们就不用费心去测试偶数了，因为 2 是唯一的偶素数。

终于介绍完啦！小伙伴们，这篇关于《使用 Go 中的并发提升运行时效率》的介绍应该让你收获多多了吧！欢迎大家收藏或分享给更多需要学习的朋友吧~golang学习网公众号也会发布Golang相关知识，快来关注吧！