multithreading - 这是Go中的惯用工作线程池吗？

coder 2023-05-01 原文

我试图用goroutines写一个简单的工作池。

我写的代码是惯用的吗？如果没有，那应该改变什么？

我希望能够将工作线程的最大数量设置为5，并在所有5个忙时都阻塞直到工作线程可用。我如何将其扩展为最多只能容纳5名 worker ？我是否会生成5个静态goroutine，并给每个work_channel？

代码:

package main

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

func worker(id string, work string, o chan string, wg *sync.WaitGroup) {
    defer wg.Done()
    sleepMs := rand.Intn(1000)
    fmt.Printf("worker '%s' received: '%s', sleep %dms\n", id, work, sleepMs)
    time.Sleep(time.Duration(sleepMs) * time.Millisecond)
    o <- work + fmt.Sprintf("-%dms", sleepMs)
}

func main() {
    var work_channel = make(chan string)
    var results_channel = make(chan string)

    // create goroutine per item in work_channel
    go func() {
        var c = 0
        var wg sync.WaitGroup
        for work := range work_channel {
            wg.Add(1)
            go worker(fmt.Sprintf("%d", c), work, results_channel, &wg)
            c++
        }
        wg.Wait()
        fmt.Println("closing results channel")
        close(results_channel)
    }()

    // add work to the work_channel
    go func() {
        for c := 'a'; c < 'z'; c++ {
            work_channel <- fmt.Sprintf("%c", c)
        }
        close(work_channel)
        fmt.Println("sent work to work_channel")
    }()

    for x := range results_channel {
        fmt.Printf("result: %s\n", x)
    }
}

最佳答案

从任何意义上讲，您的解决方案都不是工作程序goroutine池:您的代码不会限制并发的goroutine，也不会“重用” goroutines(它总是在收到新作业时启动新的goroutine)。

生产者-消费者模式

如Bruteforce MD5 Password cracker上所述，您可以使用producer-consumer pattern。您可能有一个指定的生产商goroutine，它将生成作业(要执行/计算的事情)并将其发送到作业 channel 。您可能有一个固定的使用者goroutine池(例如其中的5个)，它们会在交付作业的 channel 上循环，并且每个执行/完成接收到的作业。

当所有作业都生成并发送后，生产者goroutine可以简单地关闭jobs channel ，以适当的方式向消费者表明不再有任何作业。 channel 上的for ... range构造处理“关闭”事件并正确终止。请注意，关闭 channel 之前发送的所有作业仍将被传送。

这将导致干净的设计，将导致固定数量(但任意)的goroutine，并且将始终使用100％CPU(如果goroutine的数量大于CPU内核的数量)。它还具有的优点是它可以被“节流”与信道容量(缓冲信道)和消费者够程的数目的适当选择。

请注意，此模型并非必须具有指定的生产者goroutine。您可能也有多个goroutines来产生作业，但是当所有生产者goroutine都完成了产生作业时，您也必须也同步它们以仅关闭jobs channel ；否则，当已关闭时，尝试在jobs channel 上发送另一个作业会导致运行时 panic 。通常，生产作业的价格便宜，并且可以比执行的速度快得多，因此，这种模型可以在1个goroutine中生产它们同时消耗很多/执行它们，这在实践中是很好的。

处理结果:

如果作业有结果，则可以选择在其上可以传送结果(“发送回”)的指定结果 channel ，也可以选择在作业完成/完成时在使用者中处理结果。后者甚至可以通过具有处理结果的“回调”功能来实现。重要的是结果是可以独立处理还是需要合并(例如map-reduce框架)或汇总。

如果使用results channel ，则还需要一个goroutine来从中接收值，以防止使用者被阻塞(如果results的缓冲区将被填充，则会发生此情况)。

带results channel

与其发送简单的string值作为作业和结果，不如创建一个包装器类型，该包装器类型可以容纳任何其他信息，因此更加灵活:

type Job struct {
    Id     int
    Work   string
    Result string
}

注意Job结构也包装了结果，因此当我们发回结果时，它也包含原始的Job作为上下文-通常非常有用。还要注意，仅在 channel 上发送指针(*Job)而不是Job值是有好处的，因此无需制作Job的“无数”副本，并且Job结构值的大小也变得无关紧要。

这是生产者-消费者的样子:

我将使用2个 sync.WaitGroup 值，它们的作用如下:

var wg, wg2 sync.WaitGroup

生产者负责生成要执行的作业:

func produce(jobs chan<- *Job) {
    // Generate jobs:
    id := 0
    for c := 'a'; c <= 'z'; c++ {
        id++
        jobs <- &Job{Id: id, Work: fmt.Sprintf("%c", c)}
    }
    close(jobs)
}

完成后(不再有作业)，jobs channel 将关闭，这向消费者发出信号，表示将不再有作业到达。

请注意，produce()将jobs channel 视为仅发送，因为这是生产者仅需执行的操作:在其上发送作业(除了关闭它，但也可以在仅发送 channel 上使用)。生产者的意外接收将是编译时错误(在编译时及早发现)。

消费者的责任是在可以接收工作的情况下接收工作，并执行它们:

func consume(id int, jobs <-chan *Job, results chan<- *Job) {
    defer wg.Done()
    for job := range jobs {
        sleepMs := rand.Intn(1000)
        fmt.Printf("worker #%d received: '%s', sleep %dms\n", id, job.Work, sleepMs)
        time.Sleep(time.Duration(sleepMs) * time.Millisecond)
        job.Result = job.Work + fmt.Sprintf("-%dms", sleepMs)
        results <- job
    }
}

请注意，consume()将jobs channel 视为仅接收 channel ；消费者只需要从中接收。同样，results channel 仅发送给使用者。

还要注意，由于有多个使用者goroutine，此处无法关闭results channel ，因为只有第一个尝试关闭它的例程会成功，其他尝试会导致运行时出现 panic !在所有使用者goroutine结束之后，可以(必须)关闭results channel ，因为这样我们可以确定不会在results channel 上发送任何其他值(结果)。

我们有需要分析的结果:
func analyze(results <-chan *Job) { defer wg2.Done() for job := range results { fmt.Printf("result: %s\n", job.Result) } }

如您所见，只要结果可能出现，它也会收到结果(直到results channel 关闭)。分析仪的results channel 仅接收。

请注意 channel 类型的使用:只要足够，就仅在编译时使用单向 channel 类型来及早发现并防止错误。如果确实需要双向，则仅使用双向 channel 类型。

这就是将所有这些粘合在一起的方式:
func main() { jobs := make(chan *Job, 100) // Buffered channel results := make(chan *Job, 100) // Buffered channel // Start consumers: for i := 0; i < 5; i++ { // 5 consumers wg.Add(1) go consume(i, jobs, results) } // Start producing go produce(jobs) // Start analyzing: wg2.Add(1) go analyze(results) wg.Wait() // Wait all consumers to finish processing jobs // All jobs are processed, no more values will be sent on results: close(results) wg2.Wait() // Wait analyzer to analyze all results }

示例输出:

这是示例输出:

如您所见，在将所有作业排入队列之前，结果即将到来并得到分析:
worker #4 received: 'e', sleep 81ms worker #0 received: 'a', sleep 887ms worker #1 received: 'b', sleep 847ms worker #2 received: 'c', sleep 59ms worker #3 received: 'd', sleep 81ms worker #2 received: 'f', sleep 318ms result: c-59ms worker #4 received: 'g', sleep 425ms result: e-81ms worker #3 received: 'h', sleep 540ms result: d-81ms worker #2 received: 'i', sleep 456ms result: f-318ms worker #4 received: 'j', sleep 300ms result: g-425ms worker #3 received: 'k', sleep 694ms result: h-540ms worker #4 received: 'l', sleep 511ms result: j-300ms worker #2 received: 'm', sleep 162ms result: i-456ms worker #1 received: 'n', sleep 89ms result: b-847ms worker #0 received: 'o', sleep 728ms result: a-887ms worker #1 received: 'p', sleep 274ms result: n-89ms worker #2 received: 'q', sleep 211ms result: m-162ms worker #2 received: 'r', sleep 445ms result: q-211ms worker #1 received: 's', sleep 237ms result: p-274ms worker #3 received: 't', sleep 106ms result: k-694ms worker #4 received: 'u', sleep 495ms result: l-511ms worker #3 received: 'v', sleep 466ms result: t-106ms worker #1 received: 'w', sleep 528ms result: s-237ms worker #0 received: 'x', sleep 258ms result: o-728ms worker #2 received: 'y', sleep 47ms result: r-445ms worker #2 received: 'z', sleep 947ms result: y-47ms result: u-495ms result: x-258ms result: v-466ms result: w-528ms result: z-947ms

在Go Playground上尝试完整的应用程序。

没有results channel

如果我们不使用results channel ，则代码将大大简化，但是使用者goroutine会立即处理结果(在我们的情况下将其打印出来)。在这种情况下，我们不需要2个sync.WaitGroup值(只需要第二个就可以等待分析器完成)。

没有results channel ，完整的解决方案是这样的:
var wg sync.WaitGroup type Job struct { Id int Work string } func produce(jobs chan<- *Job) { // Generate jobs: id := 0 for c := 'a'; c <= 'z'; c++ { id++ jobs <- &Job{Id: id, Work: fmt.Sprintf("%c", c)} } close(jobs) } func consume(id int, jobs <-chan *Job) { defer wg.Done() for job := range jobs { sleepMs := rand.Intn(1000) fmt.Printf("worker #%d received: '%s', sleep %dms\n", id, job.Work, sleepMs) time.Sleep(time.Duration(sleepMs) * time.Millisecond) fmt.Printf("result: %s\n", job.Work+fmt.Sprintf("-%dms", sleepMs)) } } func main() { jobs := make(chan *Job, 100) // Buffered channel // Start consumers: for i := 0; i < 5; i++ { // 5 consumers wg.Add(1) go consume(i, jobs) } // Start producing go produce(jobs) wg.Wait() // Wait all consumers to finish processing jobs }

输出与results channel 的输出“类似”(但执行/完成顺序当然是随机的)。

在Go Playground上尝试此变体。

关于multithreading - 这是Go中的惯用工作线程池吗？，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/38170852/

惯用 multithreading code channel received go concurrency goroutine

有关multithreading - 这是Go中的惯用工作线程池吗？的更多相关文章

ruby - 如何从 ruby 中的字符串运行任意对象方法？ - 2
总的来说，我对ruby还比较陌生，我正在为我正在创建的对象编写一些rspec测试用例。许多测试用例都非常基础，我只是想确保正确填充和返回值。我想知道是否有办法使用循环结构来执行此操作。不必为我要测试的每个方法都设置一个assertEquals。例如:describeitem,"TestingtheItem"doit"willhaveanullvaluetostart"doitem=Item.new#HereIcoulddotheitem.name.shouldbe_nil#thenIcoulddoitem.category.shouldbe_nilendend但我想要一些方法来使用
ruby - 其他文件中的 Rake 任务 - 2
我试图在一个项目中使用rake，如果我把所有东西都放到Rakefile中，它会很大并且很难读取/找到东西，所以我试着将每个命名空间放在lib/rake中它自己的文件中，我添加了这个到我的rake文件的顶部:Dir['#{File.dirname(__FILE__)}/lib/rake/*.rake'].map{|f|requiref}它加载文件没问题，但没有任务。我现在只有一个.rake文件作为测试，名为“servers.rake”，它看起来像这样:namespace:serverdotask:testdoputs"test"endend所以当我运行rakeserver:testid时
ruby-on-rails - Ruby net/ldap 模块中的内存泄漏 - 2
作为我的Rails应用程序的一部分，我编写了一个小导入程序，它从我们的LDAP系统中吸取数据并将其塞入一个用户表中。不幸的是，与LDAP相关的代码在遍历我们的32K用户时泄漏了大量内存，我一直无法弄清楚如何解决这个问题。这个问题似乎在某种程度上与LDAP库有关，因为当我删除对LDAP内容的调用时，内存使用情况会很好地稳定下来。此外，不断增加的对象是Net::BER::BerIdentifiedString和Net::BER::BerIdentifiedArray，它们都是LDAP库的一部分。当我运行导入时，内存使用量最终达到超过1GB的峰值。如果问题存在，我需要找到一些方法来更正我的代
ruby-on-rails - 由于 "wkhtmltopdf"，PDFKIT 显然无法正常工作 - 2
我在从html页面生成PDF时遇到问题。我正在使用PDFkit。在安装它的过程中，我注意到我需要wkhtmltopdf。所以我也安装了它。我做了PDFkit的文档所说的一切......现在我在尝试加载PDF时遇到了这个错误。这里是错误:commandfailed:"/usr/local/bin/wkhtmltopdf""--margin-right""0.75in""--page-size""Letter""--margin-top""0.75in""--margin-bottom""0.75in""--encoding""UTF-8""--margin-left""0.75in""-
ruby-on-rails - Rails 3 中的多个路由文件 - 2
Rails2.3可以选择随时使用RouteSet#add_configuration_file添加更多路由。是否可以在Rails3项目中做同样的事情？最佳答案在config/application.rb中:config.paths.config.routes在Rails3.2(也可能是Rails3.1)中，使用:config.paths["config/routes"] 关于ruby-on-rails-Rails3中的多个路由文件，我们在StackOverflow上找到一个类似的问题
ruby-on-rails - 'compass watch' 是如何工作的/它是如何与 rails 一起使用的 - 2
我在我的项目目录中完成了compasscreate.和compassinitrails。几个问题:我已将我的.sass文件放在public/stylesheets中。这是放置它们的正确位置吗？当我运行compasswatch时，它不会自动编译这些.sass文件。我必须手动指定文件:compasswatchpublic/stylesheets/myfile.sass等。如何让它自动运行？文件ie.css、print.css和screen.css已放在stylesheets/compiled。如何在编译后不让它们重新出现的情况下删除它们？我自己编译的.sass文件编译成compiled/t
ruby-on-rails - Rails - 一个 View 中的多个模型 - 2
我需要从一个View访问多个模型。以前，我的links_controller仅用于提供以不同方式排序的链接资源。现在我想包括一个部分(我假设)显示按分数排序的顶级用户(@users=User.all.sort_by(&:score))我知道我可以将此代码插入每个链接操作并从View访问它，但这似乎不是“ruby方式”，我将需要在不久的将来访问更多模型。这可能会变得很脏，是否有针对这种情况的任何技术？注意事项:我认为我的应用程序正朝着单一格式和动态页面内容的方向发展，本质上是一个典型的网络应用程序。我知道before_filter但考虑到我希望应用程序进入的方向，这似乎很麻烦。最终从任何
ruby-on-rails - Rails 3.2.1 中 ActionMailer 中的未定义方法 'default_content_type=' - 2
我在我的项目中添加了一个系统来重置用户密码并通过电子邮件将密码发送给他，以防他忘记密码。昨天它运行良好(当我实现它时)。当我今天尝试启动服务器时，出现以下错误。=>BootingWEBrick=>Rails3.2.1applicationstartingindevelopmentonhttp://0.0.0.0:3000=>Callwith-dtodetach=>Ctrl-CtoshutdownserverExiting/Users/vinayshenoy/.rvm/gems/ruby-1.9.3-p0/gems/actionmailer-3.2.1/lib/action_mailer
ruby-on-rails - Rails 应用程序中的 Rails : How are you using application_controller. rb 是新手吗？ - 2
刚入门rails，开始慢慢理解。有人可以解释或给我一些关于在application_controller中编码的好处或时间和原因的想法吗？有哪些用例。您如何为Rails应用程序使用应用程序Controller？我不想在那里放太多代码，因为据我了解，每个请求都会调用此Controller。这是真的？最佳答案 ApplicationController实际上是您应用程序中的每个其他Controller都将从中继承的类(尽管这不是强制性的)。我同意不要用太多代码弄乱它并保持干净整洁的态度，尽管在某些情况下ApplicationContr
ruby-on-rails - form_for 中不在模型中的自定义字段 - 2
我想向我的Controller传递一个参数，它是一个简单的复选框，但我不知道如何在模型的form_for中引入它，这是我的观点:{:id=>'go_finance'}do|f|%>Transferirde:para:Entrada:"input",:placeholder=>"Quantofoiganho?"%>Saída:"output",:placeholder=>"Quantofoigasto?"%>Nota:我想做一个额外的复选框，但我该怎么做，模型中没有一个对象，而是一个要检查的对象，以便在Controller中创建一个ifelse，如果没有检查，请帮助我，非常感谢,谢谢

multithreading - 这是Go中的惯用工作线程池吗？

有关multithreading - 这是Go中的惯用工作线程池吗？的更多相关文章

随机推荐