DataMapper、互斥体和线程的奇怪错误？答案

【问题标题】：Odd bug with DataMapper, Mutexes, and Threads?DataMapper、互斥体和线程的奇怪错误？
【发布时间】：2012-08-19 02:41:39
【问题描述】：

我有一个充满 URL 的数据库，我需要定期测试其 HTTP 响应时间。我希望有许多工作线程始终为最近未测试的 URL 梳理数据库，如果找到，请对其进行测试。

当然，这可能会导致多个线程从数据库中获取相同的 URL。我不想要这个。所以，我正在尝试使用互斥锁来防止这种情况发生。我意识到在数据库级别还有其他选项（乐观锁定、悲观锁定），但我至少更愿意弄清楚为什么这不起作用。

看看我写的这段测试代码：

threads = []
mutex = Mutex.new

50.times do |i|
  threads << Thread.new do
    while true do 
      url = nil

      mutex.synchronize do
        url = URL.first(:locked_for_testing => false, :times_tested.lt => 150)
        if url
          url.locked_for_testing = true
          url.save 
        end
      end

      if url
        # simulate testing the url
        sleep 1

        url.times_tested += 1
        url.save

        mutex.synchronize do
          url.locked_for_testing = false
          url.save
        end
      end
    end

    sleep 1
  end
end

threads.each { |t| t.join }

当然这里没有真正的 URL 测试。但是应该在一天结束时发生，每个 URL 应该以等于 150 的“times_tested”结束，对吧？

（我基本上只是想确保互斥锁和工作线程心态正常工作）

但每次我运行它时，一些奇怪的 URL 会出现一些奇怪的 URL，times_tested 等于一个小得多的数字，比如 37，locked_for_testing 冻结为“true”

现在，据我的代码所知，如果任何 URL 被锁定，它将解锁。所以我不明白某些 URL 是如何以这样的方式“冻结”的。

没有例外，我尝试添加开始/确保，但它没有做任何事情。

有什么想法吗？

【问题讨论】：

里面有一个不同步的url.save，会出问题。此外，这似乎不会从使用线程中受益。
@pguardiario 你能解释一下不同步的保存是如何导致问题的吗？我的逻辑是没有其他东西应该触及那个 URL，因为它已经被锁定了。此外，这确实受益于线程，因为网络连接速度很慢，因此一台计算机可以同时测试多个 url。
是的，但是线程共享一个数据库连接。数据库操作应始终同步，因为没有理由不同步。
您可能正在耗尽连接池
@FrederickCheung 你能详细说明一下 - 我不明白

标签： ruby multithreading concurrency locking datamapper

【解决方案1】：

我会使用一个队列和一个主人来拉你想要的东西。如果您有一个主控，您可以控制访问的内容。这并不完美，但不会因为并发而崩溃，请记住，如果您没有锁定数据库，互斥锁并不能真正帮助您，因为其他东西访问了数据库。

代码完全未经测试

require 'thread'
queue = Queue.new
keep_running = true

# trap cntrl_c or something to reset keep_running
master = Thread.new do 
  while keep_running
    # check if we need some work to do
    if queue.size == 0
      urls = URL.all(:times_tested.lt => 150)
      urls.each do |u|
        queue << u.id
      end
      # keep from spinning the queue
      sleep(0.1)
    end
  end
end
workers = []
50.times do
  workers << Thread.new do
    while keep_running
      # get an id
      id = queue.shift
      url = URL.get(id)
      #do something with the url
      url.save
      sleep(0.1)
    end
  end
end
workers.each do |w|
  w.join
end

【讨论】：