Go语言的并发循环逻辑

IT行业相对于一般传统行业，发展更新速度更快，一旦停止了学习，很快就会被行业所淘汰。所以我们需要踏踏实实的不断学习，精进自己的技术，尤其是初学者。今天golang学习网给大家整理了《Go语言的并发循环逻辑》，聊聊，我们一起来看看吧！

我刚刚接触 go 中的并发性，并尝试创建一个调度 go 例程，将作业发送到监听 jobchan 通道的工作池。如果消息通过dispatchchan通道进入我的调度函数并且我的其他go例程很忙，则该消息将被附加到调度程序中的堆栈切片上，并且调度程序将在稍后当工作人员可用时再次尝试发送，和/或没有在dispatchchan 上收到更多消息。这是因为dispatchchan和jobchan是无缓冲的，并且工作人员正在运行的go例程会将其他消息附加到调度程序直到某个点，我不希望工作人员被阻塞等待调度程序并造成死锁。这是我迄今为止提出的调度程序代码：

func dispatch() {
var stack []string
acount := 0
for {
    select {
    case d := <-dispatchchan:
        stack = append(stack, d)
    case c := <-mw:
        acount = acount + c
    case jobchan <-stack[0]:
        if len(stack) > 1 {
            stack[0] = stack[len(stack)-1]
            stack = stack[:len(stack)-1]
        } else {
            stack = nil
        }
    default:
        if acount == 0 && len(stack) == 0 {
            close(jobchan)
            close(dispatchchan)
            close(mw)
            wg.Done()
            return
        }
    }
}

完整示例位于 https://play.golang.wiki/p/x6kxvnun5n7

mw通道是一个缓冲通道，其长度与worker go例程的数量相同。它充当工作池的信号量。如果工作例程正在执行 [m] 有意义的 [w] 工作，它会在 mw 通道上抛出 int 1，当它完成工作并返回到监听 jobchan 的 for 循环时，它会在 mw 上抛出 int -1。这样调度程序就知道工作池是否正在完成任何工作，或者池是否空闲。如果池空闲并且堆栈上没有更多消息，则调度程序将关闭通道并将控制权返回给主函数。

这一切都很好，但我遇到的问题是堆栈本身的长度可能为零，因此在我尝试将 stack[0] 发送到 jobchan 的情况下，如果堆栈为空，我会收到越界错误。我想要弄清楚的是如何确保当我遇到这种情况时， stack[0] 中是否有值。我不希望这种情况向 jobchan 发送空字符串。

非常感谢任何帮助。如果我应该考虑更符合个人习惯的并发模式，我很想听听。我并不是 100% 相信这个解决方案，但这是迄今为止我所取得的最远成果。

正确答案

这一切都很好，但我遇到的问题是堆栈本身的长度可能为零，因此在我尝试将 stack[0] 发送到 jobchan 的情况下，如果堆栈为空，我会收到越界错误.

我无法用您的演示链接复制它，但它是可信的，因为至少有一个 gofunc 工作人员可能已经准备好在该频道上接收。

我的输出是 msgcnt: 0，这也很容易解释，因为当 dispatch() 运行其 select 时，gofunc 可能未准备好在 jobschan 上接收。这些操作的顺序未定义。

尝试创建一个调度 go 例程，将作业发送到监听 jobchan 通道的工作池

通道不需要调度程序。通道是调度程序。

如果一条消息通过dispatchchan通道进入我的调度函数，并且我的其他go例程正忙，则该消息[...]将[...]稍后当工作人员可用时再次发送，[... ]或者dispatchchan上没有收到更多消息。

通过一些创造性的编辑，很容易将其变成接近缓冲通道定义的东西。它可以立即读取，或者可能需要一些无法立即发送的消息“limit”。您确实定义了 limit，但它没有在代码中的其他地方使用。

在任何函数中，定义不读取的变量都会导致编译时错误，例如声明了 limit 但未使用 。这种限制提高了代码质量并有助于识别打字错误。但在包范围内，您已经将未使用的 limit 定义为“全局”，从而避免了一个有用的错误 - 您没有限制任何内容。

不要使用全局变量。使用传递的参数来定义范围，因为范围的定义相当于用 go 关键字表示的功能并发。 将在本地范围内定义的相关通道传递给在包范围内定义的函数，以便您可以轻松跟踪它们的关系。并使用 directional channels 强制执行函数之间的生产者/消费者关系。稍后详细介绍。

回到“限制”，限制正在排队的作业数量是有意义的，因为所有资源都是有限的，并且接受比预期处理更多的消息需要比进程内存提供的更持久的存储。如果您觉得无论如何都没有义务满足这些请求，那么首先就不要接受“太多”的请求。

那么，dispatchchan 和 dispatch() 有什么函数呢？在处理之前存储有限数量的待处理请求（如果有），然后将它们发送给下一个可用的工作人员？这正是缓冲通道的用途。

循环逻辑

谁“知道”你的程序何时完成？ main() 提供初始输入，但您在 `dispatch() 中关闭所有 3 个通道：

close(jobchan)
            close(dispatchchan)
            close(mw)

您的工作人员会写入自己的作业队列，因此只有当工作人员完成写入后，才能关闭传入的作业队列。然而，个别工作人员也不知道何时关闭作业队列，因为其他工作人员正在向其写入数据。 没有人知道你的算法何时完成。这就是你的循环逻辑。

mw通道是一个缓冲通道，其长度与worker go例程的数量相同。它充当工作池的信号量。

这里存在竞争条件。考虑所有 n 工作人员刚刚收到最后一份 n 作业的情况。他们每个人都从 jobschan 读取数据，并检查 ok 的值。 disptatcher 继续运行其 select。目前没有人向 dispatchchan 写入或从 jobschan 读取，因此 default 大小写会立即匹配。 len(stack) 是 0 并且没有当前的 job，因此 dispatcher 关闭包括 mw 在内的所有通道。此后的某个时刻，工作人员尝试写入已关闭的通道并出现恐慌。

所以我终于准备好提供一些代码了，但我还有一个问题：我没有明确的问题陈述来编写代码。

我刚刚接触 go 中的并发性，并尝试创建一个调度 go 例程，将作业发送到监听 jobchan 通道的工作池。

goroutine 之间的通道就像同步齿轮的齿。但齿轮转动的目的是什么？你并不是要计时，也不是要制造一个发条玩具。你的齿轮可以转动，但成功是什么样子的呢？他们的转变？

让我们尝试为通道定义一个更具体的用例：给定一组任意长的持续时间作为标准输入*上的字符串，在 n 工作人员中休眠那么多秒。因此，我们实际上有一个要返回的结果，我们会说每个工作线程都会返回运行持续时间的开始时间和结束时间。

• 为了使其能够在演示中运行，我将使用硬编码字节缓冲区模拟标准输入。

package main

import (
    "bufio"
    "bytes"
    "fmt"
    "os"
    "strings"
    "sync"
    "time"
)

type sleepresult struct {
    worker_id int
    duration  time.duration
    start     time.time
    end       time.time
}

func main() {
    var num_workers = 2
    workchan := make(chan time.duration)
    resultschan := make(chan sleepresult)
    var wg sync.waitgroup
    var resultswg sync.waitgroup
    resultswg.add(1)
    go results(&resultswg, resultschan)
    for i := 0; i < num_workers; i++ {
        wg.add(1)
        go worker(i, &wg, workchan, resultschan)
    }
    // playground doesn't have stdin
    var input = bytes.newbufferstring(
        strings.join([]string{
            "3ms",
            "1 seconds",
            "3600ms",
            "300 ms",
            "5s",
            "0.05min"}, "\n") + "\n")

    var scanner = bufio.newscanner(input)
    for scanner.scan() {
        text := scanner.text()
        if dur, err := time.parseduration(text); err != nil {
            fmt.fprintln(os.stderr, "invalid duration", text)
        } else {
            workchan <- dur
        }
    }
    close(workchan) // we know when our inputs are done
    wg.wait()       // and when our jobs are done
    close(resultschan)
    resultswg.wait()
}

func results(wg *sync.waitgroup, resultschan <-chan sleepresult) {
    for res := range resultschan {
        fmt.printf("worker %d: %s : %s => %s\n",
            res.worker_id, res.duration,
            res.start.format(time.rfc3339nano), res.end.format(time.rfc3339nano))
    }
    wg.done()
}

func worker(id int, wg *sync.waitgroup, jobchan <-chan time.duration, resultschan chan<- sleepresult) {
    var res = sleepresult{worker_id: id}
    for dur := range jobchan {
        res.duration = dur
        res.start = time.now()
        time.sleep(res.duration)
        res.end = time.now()
        resultschan <- res
    }
    wg.done()
}

这里我使用 2 个等待组，一组用于工作人员，一组用于结果。这确保我在 main() 结束之前写完所有结果。我通过让每个函数一次只做一件事来保持我的函数简单：main 读取输入，解析它们的持续时间，并将它们发送给下一个工作人员。 results 函数收集结果并将其打印到标准输出。工作线程负责睡眠、读取 jobchan 并写入 resultschan。

workchan 可以缓冲（也可以不缓冲，如本例所示）；这并不重要，因为输入将以可处理的速率读取。我们可以缓冲尽可能多的输入，但我们不能缓冲无限量。我已将通道大小设置为 1e6 一样大 - 但一百万远小于无限。对于我的用例，我根本不需要进行任何缓冲。

main 知道输入何时完成，并可以关闭 jobschan。 main 还知道作业何时完成 (wg.wait()) 并可以关闭结果通道。关闭这些通道对于 worker 和 results goroutine 来说是一个重要信号 - 它们可以区分空通道和保证不添加任何新通道的通道。

for job := range jobchan {...} 是更详细的简写：

for {
  job, ok :=  <- jobchan
  if !ok {
    wg.Done()
    return
  }
  ...
}

请注意，此代码创建了 2 个工作线程，但也可能创建 20 个或 2000 个工作线程，甚至 1 个。无论池中有多少工作线程，程序都会运行。它可以处理任何数量的输入（尽管无休止的输入当然会导致无休止的程序）。它不会创建输出到输入的循环。如果您的用例需要工作来创造更多工作，那么这是一个更具挑战性的场景，通常可以通过仔细规划来避免。

我希望这能为您提供一些关于如何在 go 应用程序中更好地使用并发性的好主意。

https://play.golang.wiki/p/cZuI9YXypxI

到这里，我们也就讲完了《Go语言的并发循环逻辑》的内容了。个人认为，基础知识的学习和巩固，是为了更好的将其运用到项目中，欢迎关注golang学习网公众号，带你了解更多关于的知识点！