マルチスレッド デザインパターン(Producer-Consumer パターン) Ruby編

マルチスレッドデザインパターンであるProducer-Consumerパターンについてまとめる。

このパターンはスレッド間で処理の待ち合わせを行いながら処理を実行させていきたい場合に利用する。

【登場人物】
・キューにデータを入れるスレッド
・キューからデータを取り出すスレッド

【条件】
・キューが空であるならば、データの取得処理は待たなければならない
・キューのサイズの上限まできたならば、データの入れ込み処理は待たなければならない

なぜキューを使うかというと、ProducerとConsumerの処理スピードが違うとスループットが落ちるためである。そこで中継地点としてキューをおき、そこにデータを保持させ、処理スピードの違いを吸収させる。

今回はRubyを利用する。
Javaであればwait、notifyAllを使う場面ではそれぞれ、ConditionVariableのwait、broadcastを利用している。また、今回は待機させる全スレッドを再稼働させるためsignal(notify)ではなく、broadcastを使う方針をとる。
(参考)
http://www.techscore.com/tech/Java/JavaSE/Thread/5-2/

#!/usr/bin/ruby

require 'thread'

class Table
  def initialize()
    @queues = Array.new
    @queue_max = 5

    @mutex = Mutex.new
    @cv = ConditionVariable.new
  end

  def push(num)
    @mutex.synchronize do
      while @queues.size() >= @queue_max
        @cv.wait(@mutex)
      end

      @queues.push(num)

      @cv.broadcast
    end
  end

  def take()
    @mutex.synchronize do
      while @queues.size() == 0
        @cv.wait(@mutex)
      end

      queue = @queues.shift()

      @cv.broadcast
    end
  end
end


class Push_Take
  def initialize(table)
    @table = table
  end

  def start()
    @push_thread = Thread.start do
      10.times do |num|
        @table.push(num)
        puts "push"
      end
    end

    @take_thread = Thread.start do
      10.times do
        @table.take()
        puts "take"
        sleep 0.5
      end
    end

    @push_thread.join
    @take_thread.join
  end
end


table = Table.new
push_take = Push_Take.new(table)
push_take.start()

スレッドパターン以前の並行処理プログラミングの基礎はここを参考してほしい。