Queueを使ったワーカースレッド

Queueを使うとワーカースレッドが簡単に作れます。

• ワーカースレッド(=仕事をするスレッド)は、キューから順番に仕事を取り出して実行するスレッドです。
  • これとは別に定期的に仕事を積むスレッドがいて、
  • ワーカースレッドは↑から仕事が積まれた場合にそれを取り出して実行します。
  • 仕事がなければ、積まれるまで待つという動作をします。
  • ここで、キューへの仕事の追加と、キューからの仕事の取得は、別のスレッドで行われるため、同期制御が必要になります。
• Queueを使うと、仕事を積む人とワーカースレッドの同期制御が内部で隠蔽されるため、外側での同期化なしにワーカースレッドを作ることができます。
  • QueueのAPIはpopとpushがあります。(他にもあるけど)
  • Queue#popで要素を取り出します。このとき、queueが空の時、呼出元のスレッドは停止(ブロック)されます。
  • Queue#pushで要素を追加します。このとき、待っているスレッドがいれば実行を再開させます。
  • 仕事をするスレッドがpushで仕事を積み、ワーカースレッドがpopで仕事を取り出すようにすれば、「仕事が積まれた場合にそれを取り出して実行。仕事がなければ、積まれるまで待つ」という動作を実現できます。

以下はQueueを使ったワーカースレッドのサンプルです。

require 'thread'

# キュー
q = Queue.new

# ワーカースレッド。
# キューから仕事を取り出して実行する。
# 仕事がなければ待つ。
workers = []
3.times { |j|
  workers << Thread.start {
    name = "thread-" << j.to_s
    loop {
      q.pop.call( name ) # キューから仕事を取り出して実行する。
      sleep rand * 0.1
    }
  }
}

# 仕事を作るスレッド。
makers = []
3.times { |j| # 3人で
  makers << Thread.start {
    5.times {|i| # 5個づつ
      q.push Proc.new {|name|  # キューに仕事を積む。
        puts name + " : " + i.to_s
      }
      sleep rand * 0.1
    }
  }
}

sleep 10 # ほんとはちゃんと待たないといけない。

実行結果です。

thread-0 : 0
thread-1 : 0
thread-2 : 0
thread-2 : 1
thread-1 : 1
thread-0 : 1
thread-2 : 2
thread-0 : 2
thread-1 : 2
thread-2 : 3
thread-2 : 3
thread-1 : 4
thread-2 : 4
thread-0 : 3
thread-1 : 4