草庐IT

go - 如何处理可以无阻塞增长的队列

coder 2024-07-07 原文

如果队列可以从处理函数本身增长，我正在尝试了解如何在 Go 中处理队列。请参见下面的代码。

在这个伪代码中，我想将我创建的处理程序数量限制为 10。因此我创建了 10 个处理队列的处理程序。然后我用一个 url 开始排队。

我的问题是，根据文档， channel 的 sender 将阻塞，直到接收者接收到数据。在下面的代码中，每个进程都是一个处理新 url 的接收器。然而，很容易看出，如果一个进程向队列发送 11 个链接，它将阻塞，直到所有接收者都处理完这些新链接。如果这些接收者每个都有 1 个链接，那么它们也会在将新的 1 个链接发送到队列时阻塞。由于每个人都被阻止，所以什么都没有完成。

我想知道 go 的一般解决方案是什么，用于处理可以从进程本身增长的队列。请注意，我认为我可以通过锁定名为 queue 的数组来执行此操作，但我正在尝试了解如何使用 channel 完成此操作。

var queue = make(chan string)

func process(){
    for currentURL := range queue {
        links, _ := ... // some http call that gets links from a url
        for _, link := links {
            queue <- link
        }
    }
}

func main () {
   for i :=0; i < 10; i++ {
        go process()
   }

   queue <- "https://stackoverflow.com"
   ...
   // block until receive some quit message
   <-quit 
}

最佳答案

您可以使用的一种简单方法是将用于将链接添加到 channel 的代码移动到它自己的 go 例程中。 这样，您的主要处理可以继续，而阻塞的 channel 写入将阻塞一个单独的 go 例程。

func process(){
    for currentURL := range queue {
        links, _ := ... // some http call that gets links from a url
        for _, link := links {
            l := link // this is important! ...
            // the loop will re-set the value of link before the go routine is started

            go func(l) {
                queue <- link // we'll be blocked here...
                // but the "parent" routine can still iterate through the channel
                // which in turn un-blocks the write
            }(l)
        }
    }
}

使用信号量示例编辑以限制 go 例程:

func main () {
    maxWorkers := 5000
    sem := semaphore.NewWeighted(int64(maxWorkers))
    ctx := context.TODO()
    for i :=0; i < 10; i++ {
        go process(ctx)
    }

    queue <- "https://stackoverflow.com"
    // block until receive some quit message
    <-quit 
}

func process(ctx context.Context){
    for currentURL := range queue {
        links, _ := ... // some http call that gets links from a url
        for _, link := links {
            l := link // this is important! ...
            // the loop will re-set the value of link before the go routine is started

            // acquire a go routine...
            // if we are at the routine limit, this line will block until one becomes available
            sem.Acquire(ctx, 1)
            go func(l) {
                defer sem.Release(1)
                queue <- link // we'll be blocked here...
                // but the "parent" routine can still iterate through the channel
                // which in turn un-blocks the write
            }(l)
        }
    }
}

尽管此选项最终可能导致死锁...假设所有 go 例程都已声明，父循环可能会锁定在 sem.Acquire 上。这将导致子例程永远不会添加到 channel ，因此永远不会执行延迟的 sem.Release。在我的脑海中，我正在努力想出一个很好的方法来处理这个问题。也许是外部内存队列而不是 channel ？

关于go - 如何处理可以无阻塞增长的队列，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/51911766/

无阻何处codequeuesectiongochannelgoroutine

有关go - 如何处理可以无阻塞增长的队列的更多相关文章

  1. ruby - 为什么我可以在 Ruby 中使用 Object#send 访问私有(private)/ protected 方法？ - 2

    类classAprivatedeffooputs:fooendpublicdefbarputs:barendprivatedefzimputs:zimendprotecteddefdibputs:dibendendA的实例a=A.new测试a.foorescueputs:faila.barrescueputs:faila.zimrescueputs:faila.dibrescueputs:faila.gazrescueputs:fail测试输出failbarfailfailfail.发送测试[:foo,:bar,:zim,:dib,:gaz].each{|m|a.send(m)resc

  2. ruby - 使用 Vim Rails，您可以创建一个新的迁移文件并一次性打开它吗？ - 2

    使用带有Rails插件的vim，您可以创建一个迁移文件，然后一次性打开该文件吗？textmate也可以这样吗？ 最佳答案 你可以使用rails.vim然后做类似的事情::Rgeneratemigratonadd_foo_to_bar插件将打开迁移生成的文件，这正是您想要的。我不能代表textmate。 关于ruby-使用VimRails，您可以创建一个新的迁移文件并一次性打开它吗？，我们在StackOverflow上找到一个类似的问题： https://sta

  3. ruby - 我可以使用 Ruby 从 CSV 中删除列吗？ - 2

    查看Ruby的CSV库的文档，我非常确定这是可能且简单的。我只需要使用Ruby删除CSV文件的前三列，但我没有成功运行它。 最佳答案 csv_table=CSV.read(file_path_in,:headers=>true)csv_table.delete("header_name")csv_table.to_csv#=>ThenewCSVinstringformat检查CSV::Table文档:http://ruby-doc.org/stdlib-1.9.2/libdoc/csv/rdoc/CSV/Table.html

  4. ruby - 我可以使用 aws-sdk-ruby 在 AWS S3 上使用事务性文件删除/上传吗？ - 2

    我发现ActiveRecord::Base.transaction在复杂方法中非常有效。我想知道是否可以在如下事务中从AWSS3上传/删除文件:S3Object.transactiondo#writeintofiles#raiseanexceptionend引发异常后，每个操作都应在S3上回滚。S3Object这可能吗？？ 最佳答案 虽然S3API具有批量删除功能，但它不支持事务，因为每个删除操作都可以独立于其他操作成功/失败。该API不提供任何批量上传功能(通过PUT或POST)，因此每个上传操作都是通过一个独立的API调用完成的

  5. ruby-on-rails - Enumerator.new 如何处理已通过的 block ？ - 2

    我在理解Enumerator.new方法的工作原理时遇到了一些困难。假设文档中的示例:fib=Enumerator.newdo|y|a=b=1loopdoy[1,1,2,3,5,8,13,21,34,55]循环中断条件在哪里，它如何知道循环应该迭代多少次(因为它没有任何明确的中断条件并且看起来像无限循环)？ 最佳答案 Enumerator使用Fibers在内部。您的示例等效于:require'fiber'fiber=Fiber.newdoa=b=1loopdoFiber.yieldaa,b=b,a+bendend10.times.m

  6. ruby - 有人可以帮助解释类创建的 post_initialize 回调吗 (Sandi Metz) - 2

    我正在阅读SandiMetz的POODR，并且遇到了一个我不太了解的编码原则。这是代码:classBicycleattr_reader:size,:chain,:tire_sizedefinitialize(args={})@size=args[:size]||1@chain=args[:chain]||2@tire_size=args[:tire_size]||3post_initialize(args)endendclassMountainBike此代码将为其各自的属性输出1,2,3,4,5。我不明白的是查找方法。当一辆山地自行车被实例化时，因为它没有自己的initialize方法

  7. ruby - 是否可以覆盖 gemfile 进行本地开发？ - 2

    我们的git存储库中目前有一个Gemfile。但是，有一个gem我只在我的环境中本地使用(我的团队不使用它)。为了使用它，我必须将它添加到我们的Gemfile中，但每次我checkout到我们的master/dev主分支时，由于与跟踪的gemfile冲突，我必须删除它。我想要的是类似Gemfile.local的东西，它将继承从Gemfile导入的gems，但也允许在那里导入新的gems以供使用只有我的机器。此文件将在.gitignore中被忽略。这可能吗？ 最佳答案 设置BUNDLE_GEMFILE环境变量:BUNDLE_GEMFI

  8. ruby - 我可以将我的 README.textile 以正确的格式放入我的 RDoc 中吗？ - 2

    我喜欢使用Textile或Markdown为我的项目编写自述文件，但是当我生成RDoc时，自述文件被解释为RDoc并且看起来非常糟糕。有没有办法让RDoc通过RedCloth或BlueCloth而不是它自己的格式化程序运行文件？它可以配置为自动检测文件后缀的格式吗？(例如README.textile通过RedCloth运行，但README.mdown通过BlueCloth运行) 最佳答案 使用YARD直接代替RDoc将允许您包含Textile或Markdown文件，只要它们的文件后缀是合理的。我经常使用类似于以下Rake任务的东西:

  9. ruby - 一个 YAML 对象可以引用另一个吗？ - 2

    我想让一个yaml对象引用另一个，如下所示:intro:"Hello,dearuser."registration:$introThanksforregistering!new_message:$introYouhaveanewmessage!上面的语法只是它如何工作的一个例子(这也是它在thiscpanmodule中的工作方式。)我正在使用标准的ruby​​yaml解析器。这可能吗？ 最佳答案 一些yaml对象确实引用了其他对象:irb>require'yaml'#=>trueirb>str="hello"#=>"hello"ir

  10. ruby - 可以通过多少种方法将方法添加到 ruby​​ 对象？ - 2

    当谈到运行时自省(introspection)和动态代码生成时，我认为ruby​​没有任何竞争对手，可能除了一些lisp方言。前几天，我正在做一些代码练习来探索ruby​​的动态功能，我开始想知道如何向现有对象添加方法。以下是我能想到的3种方法:obj=Object.new#addamethoddirectlydefobj.new_method...end#addamethodindirectlywiththesingletonclassclass这只是冰山一角，因为我还没有探索instance_eval、module_eval和define_method的各种组合。是否有在线/离线资

随机推荐