草庐IT

go - 重新插入 channel 导致死锁

coder 2023-07-02 原文

我有一个稳定的“工作”入站流，我将它们输入到一个无缓冲的 channel 中。我有一个 for range 循环来遍历项目并处理它们。如果处理项目失败，我会将项目重新插入 channel ，以便稍后重试。

问题是当我将项目重新插入 channel 时 - 它死锁了。我明白为什么会这样:处理器在尝试发送时没有从 channel 读取数据，因此发送会永远阻塞。但我想不出解决问题的模式。有人可以协助找到解决方案吗？

这是一个显示我的问题的简单示例代码 (https://play.golang.org/p/N_-jWL5aOCo):

package main

import (
    "fmt"
    "time"
)

type Job struct {
    ID       int
    Attempts int
}

func main() {
    ch := make(chan *Job)
    go fetchJobs(ch)

    for job := range ch {
        if success := processJob(job); !success {
            ch <- job
        }
    }
}

func processJob(job *Job) bool {
    job.Attempts++
    fmt.Printf("Processing job %+v\n", job)

    // Simulate work.
    time.Sleep(time.Millisecond * 500)

    // Simulate failure on some jobs (IDs 10 to 19, 30 to 39, etc.)
    if job.ID%20 >= 10 && job.Attempts == 1 {
        return false
    }

    return true
}

func fetchJobs(ch chan *Job) {
    for i := 0; ; i++ {
        ch <- &Job{ID: i}
    }
}

最佳答案

最简单的解决方案是使用一个新的goroutine把它放回Go:

if success := processJob(job); !success {
    go func() { ch <- job }()
}

如果您想避免为此使用新的 goroutine，另一种解决方案是“存储”失败的作业。最简单的存储可能是 slice 。如果作业处理失败，则将作业附加到失败的作业。

生产者在获取新作业之前(或之后，取决于您希望将失败的作业重新排队的“速度”有多快)可以检查是否有失败的作业，如果有，则将其中的一些(或全部)放入队列。当然，必须同步访问这个失败的作业存储。

另请注意，您不应无条件地重新排队失败的作业，因为如果错误是永久性的，它们将永远不会完成，可能会阻塞您的整个系统。一个简单的解决方法是仅在重试计数器小于限制时才重新排队。

虽然如果您有一个无缓冲的作业 channel 和一个生产者和消费者，重新排队可能是一个不必要的并发症。您也可以重试几次失败的作业，如果在某个重试或时间限制内无法成功，则将其处理为可撤销。

关于go - 重新插入 channel 导致死锁，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/56077232/

channelgosectionjob

有关go - 重新插入 channel 导致死锁的更多相关文章

  1. ruby - 如何在续集中重新加载表模式？ - 2

    鉴于我有以下迁移:Sequel.migrationdoupdoalter_table:usersdoadd_column:is_admin,:default=>falseend#SequelrunsaDESCRIBEtablestatement,whenthemodelisloaded.#Atthispoint,itdoesnotknowthatusershaveais_adminflag.#Soitfails.@user=User.find(:email=>"admin@fancy-startup.example")@user.is_admin=true@user.save!ende

  2. ruby-on-rails - active_admin 目录中的常量警告重新声明 - 2

    我正在使用active_admin，我在Rails3应用程序的应用程序中有一个目录管理，其中包含模型和页面的声明。时不时地我也有一个类，当那个类有一个常量时，就像这样:classFooBAR="bar"end然后，我在每个必须在我的Rails应用程序中重新加载一些代码的请求中收到此警告:/Users/pupeno/helloworld/app/admin/billing.rb:12:warning:alreadyinitializedconstantBAR知道发生了什么以及如何避免这些警告吗？ 最佳答案 在纯Ruby中:classA

  3. ruby - 在 Ruby 中重新分配常量时抛出异常？ - 2

    我早就知道Ruby中的“常量”(即大写的变量名)不是真正常量。与其他编程语言一样，对对象的引用是唯一存储在变量/常量中的东西。(侧边栏:Ruby确实具有“卡住”引用对象不被修改的功能，据我所知，许多其他语言都没有提供这种功能。)所以这是我的问题:当您将一个值重新分配给常量时，您会收到如下警告:>>FOO='bar'=>"bar">>FOO='baz'(irb):2:warning:alreadyinitializedconstantFOO=>"baz"有没有办法强制Ruby抛出异常而不是打印警告？很难弄清楚为什么有时会发生重新分配。 最佳答案

  4. Ruby 守护进程导致 ActiveRecord 记录器 IOError - 2

    我目前正在用Ruby编写一个项目，它使用ActiveRecordgem进行数据库交互，我正在尝试使用ActiveRecord::Base.logger记录所有数据库事件具有以下代码的属性ActiveRecord::Base.logger=Logger.new(File.open('logs/database.log','a'))这适用于迁移等(出于某种原因似乎需要启用日志记录，因为它在禁用时会出现NilClass错误)但是当我尝试运行包含调用ActiveRecord对象的线程守护程序的项目时脚本失败并出现以下错误/System/Library/Frameworks/Ruby.frame

  5. ruby - 将全局 $stdout 重新分配给控制台 - ruby - 2

    我正在尝试将$stdout设置为临时写入一个文件，然后返回到一个文件。test.rb:old_stdout=$stdout$stdout.reopen("mytestfile.out",'w+')puts"thisgoesinmytestfile"$stdout=old_stdoutputs"thisshouldbeontheconsole"$stdout.reopen("mytestfile1.out",'w+')puts"thisgoesinmytestfile1:"$stdout=old_stdoutputs"thisshouldbebackontheconsole"这是输出。r

  6. ruby - 从另一个私有(private)方法中使用 self.xxx() 调用私有(private)方法 xxx，导致错误 "private method ` xxx' called” - 2

    我正在尝试获得良好的Ruby编码风格。为防止意外调用具有相同名称的局部变量，我总是在适当的地方使用self.。但是现在我偶然发现了这个:classMyClass上面的代码导致错误privatemethodsanitize_namecalled但是当删除self.并仅使用sanitize_name时，它会起作用。这是为什么？ 最佳答案 发生这种情况是因为无法使用显式接收器调用私有(private)方法，并且说self.sanitize_name是显式指定应该接收sanitize_name的对象(self)，而不是依赖于隐式接收器(也是

  7. ruby - 如何在 Ruby 字符串中插入项目符号字符？ - 2

    我正在尝试创建一个带有项目符号字符的Ruby1.9.3字符串。str="•"+"helloworld"但是，当我输入它时，我收到有关非ASCII字符的语法错误。我该怎么做？ 最佳答案 你可以把Unicode字符放在那里。str="\u2022"+"helloworld" 关于ruby-如何在Ruby字符串中插入项目符号字符？，我们在StackOverflow上找到一个类似的问题： https://stackoverflow.com/questions/1195

  8. ruby - Faye WebSocket，关闭处理程序被触发后重新连接到套接字 - 2

    我有一个super简单的脚本，它几乎包含了FayeWebSocketGitHub页面上用于处理关闭连接的内容:ws=Faye::WebSocket::Client.new(url,nil,:headers=>headers)ws.on:opendo|event|p[:open]#sendpingcommand#sendtestcommand#ws.send({command:'test'}.to_json)endws.on:messagedo|event|#hereistheentrypointfordatacomingfromtheserver.pJSON.parse(event.d

  9. ruby - 在 ruby​​ 中使用自动创建插入数组 - 2

    我想知道是否可以通过自动创建数组来插入数组，如果数组不存在的话，就像在PHP中一样:$toto[]='titi';如果尚未定义$toto，它将创建数组并将“titi”压入。如果已经存在，它只会推送。在Ruby中我必须这样做:toto||=[]toto.push('titi')可以一行完成吗？因为如果我有一个循环，它会测试“||=”，除了第一次:Person.all.eachdo|person|toto||=[]#with1billionofperson,thislineisuseless999999999times...toto.push(person.name)你有更好的解决方案吗？

  10. ruby-on-rails - 在不重新查询数据库的情况下重新排序 Rails 中的事件记录？ - 2

    例如，假设我有一个名为Products的模型，并且在ProductsController中，我有以下代码用于product_listView以显示已排序的产品。@products=Product.order(params[:order_by])让我们想象一下，在product_listView中，用户可以使用下拉菜单按价格、评级、重量等进行排序。数据库中的产品不会经常更改。我很难理解的是，每次用户选择新的order_by过滤器时，rails是否必须查询，或者rails是否能够以某种方式缓存事件记录以在服务器端重新排序？有没有一种方法可以编写它，以便在用户排序时rails不会重新查询结果

随机推荐