Ruby同步：如何使线程以正确的顺序一个接一个地工作？

Question

我的问题是我不知道如何使用 Ruby 同步多个线程。任务是创建六个线程并立即启动它们。他们都应该按照我需要它工作的顺序一个接一个地做一些工作（例如puts "Thread 1" Hi"）。

我尝试过使用 Mutex、Monitor 和 Condition Variable，但它们都以随机顺序工作。谁能解释一下如何实现我的目标？

经过一段时间与 Mutex 和 Condition Variable 的斗争后，我已经实现了我的目标。 这段代码有点乱，我故意不使用循环来“更清晰的视图”。

cv = ConditionVariable.new
mutex = Mutex.new

mutex2 = Mutex.new
cv2 = ConditionVariable.new

mutex3 = Mutex.new
cv3 = ConditionVariable.new

mutex4 = Mutex.new
cv4 = ConditionVariable.new

mutex5 = Mutex.new
cv5 = ConditionVariable.new

mutex6 = Mutex.new
cv6 = ConditionVariable.new



Thread.new do
  mutex.synchronize {
    puts 'First: Hi'
    cv.wait(mutex)
    puts 'First: Bye'
    #cv.wait(mutex)
    cv.signal
    puts 'First: One more time'
  }

end

Thread.new do
  mutex.synchronize {
    puts 'Second: Hi'
    cv.signal
    cv.wait(mutex)
    puts 'Second:Bye'
    cv.signal
  }

  mutex2.synchronize {
    puts  'Second: Starting third'
    cv2.signal

  }
end

Thread.new do
  mutex2.synchronize {
    cv2.wait(mutex2)
    puts 'Third: Hi'
  }

  mutex3.synchronize {
    puts 'Third: Starting forth'
    cv3.signal
  }
end

Thread.new do
  mutex3.synchronize {
    cv3.wait(mutex3)
    puts 'Forth: Hi'
  }

  mutex4.synchronize {
    puts 'Forth: Starting fifth'
    cv4.signal
  }
end

Thread.new do
  mutex4.synchronize {
    cv4.wait(mutex4)
    puts 'Fifth: Hi'
  }

  mutex5.synchronize {
    puts 'Fifth: Starting sixth'
    cv5.signal
  }
end

Thread.new {
  mutex5.synchronize {
    cv5.wait(mutex5)
    puts 'Sixth:Hi'
  }
}

sleep 2

Answer 1

使用队列作为 PV 信号量

您可以滥用Queue，像使用传统的PV Semaphore 一样使用它。为此，您需要创建一个 Queue 实例：

require 'thread'
...
sem = Queue.new

当一个线程需要等待时，它会调用Queue#deq：

# waiting thread
sem.deq

当某个其他线程想要解除对等待线程的阻塞时，它会将一些东西（任何东西）推送到队列中：

# another thread that wants to unblock the waiting thread
sem.enq :go

工人阶级

这是一个使用 Queue 来同步其启动和停止的工作类：

class Worker

  def initialize(worker_number)  
    @start = Queue.new
    Thread.new do
      @start.deq
      puts "Thread #{worker_number}"
      @when_done.call
    end
  end

  def start
    @start.enq :start
  end

  def when_done(&block)
    @when_done = block
  end

end

在构造时，工作线程会创建一个线程，但该线程随后会在 @start 队列上等待。直到#start 被调用，线程才会解除阻塞。

完成后，线程将执行调用到#when_done 的块。稍后我们将看到它是如何使用的。

创造工人

首先，让我们确保如果任何线程引发异常，我们会发现它：

Thread.abort_on_exception = true

我们需要六名工人：

workers = (1..6).map { |i| Worker.new(i) }

告诉每个工人完成后该做什么

这就是#when_done 发挥作用的地方：

workers.each_cons(2) do |w1, w2|
  w1.when_done { w2.start }
end

这需要每对工人轮流进行。每个工人除了最后一个被告知，当它完成时，它应该在它之后启动工人。那只剩下最后一个工人了。当它完成时，我们希望它通知这个线程：

all_done = Queue.new
workers.last.when_done { all_done.enq :done }

我们走吧！

现在剩下的就是启动第一个线程：

workers.first.start

并等待最后一个线程完成：

all_done.deq

输出：

Thread 1
Thread 2
Thread 3
Thread 4
Thread 5
Thread 6

Answer 2

如果您刚开始使用线程，您可能想尝试一些简单的方法。让第一个线程休眠 1 秒，第二个线程休眠 2 秒，第三个线程休眠 3 秒，依此类推：

$stdout.sync = true

threads = []
(1..6).each do |i|
  threads << Thread.new {
    sleep i
    puts "Hi from thread #{i}"
  }
end

threads.each(&:join)

输出（需要 6 秒，因为线程并行运行）：

Hi from thread 1
Hi from thread 2
Hi from thread 3
Hi from thread 4
Hi from thread 5
Hi from thread 6

Answer 3

您可以为每个人分配一个数字，这将表示它在队列中的位置，并检查它以查看轮到谁了：

class QueuedWorker

  def initialize(mutex, condition_variable, my_turn)
    @mutex = mutex
    @my_turn = my_turn
    @condition_variable = condition_variable
  end

  def self.turn
    @turn ||= 0
  end

  def self.done
    @turn = turn + 1
  end

  def run
    loop do
      @mutex.synchronize do
        if QueuedWorker.turn == @my_turn
          # do actual work
          QueuedWorker.done
          @condition_variable.signal
          return
        end
        @condition_variable.signal
        @condition_variable.wait(@mutex)
      end
    end
  end
end

mutex = Mutex.new
cv = ConditionVariable.new

(0..10).each do |i|
  Thread.new do
    QueueWorker.new(mutex, cv, i).run
  end
end

话虽如此，实现很尴尬，因为线程专门不是为串行工作而构建的。如果您需要连续工作，请在单个线程中完成。