如何使用条件变量？

Question

关于 Ruby 中的条件变量的资源并不多，但大多数都是错误的。就像ruby​​-doc一样，教程在这里或在这里发布- 他们都可能遇到死锁。

sleep我们可以通过按给定顺序启动线程并可能在中间放置一些线程来强制同步来解决问题。但这只是推迟了真正的问题。

我将代码重写为经典的生产者-消费者问题：

require 'thread'
queue = []
mutex = Mutex.new
resource = ConditionVariable.new
threads = []

threads << Thread.new do
  5.times do |i|
    mutex.synchronize do 
      resource.wait(mutex)
      value = queue.pop
      print "consumed #{value}\n"
    end
  end
end

threads << Thread.new do
  5.times do |i|
    mutex.synchronize do
      queue << i
      print "#{i} produced\n"
      resource.signal
    end
    sleep(1) #simulate expense
  end
end

threads.each(&:join)

有时你会得到这个（但并非总是如此）：

0 produced
1 produced
consumed 0
2 produced
consumed 1
3 produced
consumed 2
4 produced
consumed 3
producer-consumer.rb:30:in `join': deadlock detected (fatal)
        from producer-consumer.rb:30:in `each'
        from producer-consumer.rb:30:in `<main>'

正确的解决方案是什么？

Answer 1

这是更强大的解决方案，具有多个消费者和生产者以及 MonitorMixin 的使用，具有特殊MonitorMixin的方法ConditionVariable和方法wait_while()wait_until()

require 'monitor'

queue = []
queue.extend(MonitorMixin)
cond = queue.new_cond
consumers, producers = [], []

for i in 0..5
  consumers << Thread.start(i) do |i|
      print "consumer start #{i}\n"
      while (producers.any?(&:alive?) || !queue.empty?)
        queue.synchronize do
        cond.wait_while { queue.empty? }
        print "consumer #{i}: #{queue.shift}\n"
      end
      sleep(0.2) #simulate expense
    end
  end
end

for i in 0..3
  producers << Thread.start(i) do |i|
    id = (65+i).chr
    for j in 0..10 do
      queue.synchronize do
        item = "#{j} #{id}"
        queue << item
        print "producer #{id}: produced #{item}\n"
        j += 1
        cond.broadcast
      end
      sleep(0.1) #simulate expense
    end
  end
end

sleep 0.1 while producers.any?(&:alive?)
sleep 0.1 while consumers.any?(&:alive?)

print "queue size #{queue.size}\n"