Rubyのマルチスレッド処理【プログラミング初心者向け教材】

たくさんの処理を効率よく行うための手法として並列処理があります。

処理の単位をスレッドと呼び、並列で処理することをマルチスレッドと呼びます。

Rubyではマルチスレッド処理を行うための機能が標準で提供されています。

本記事では、Rubyのマルチスレッド処理について解説します。

プログラミング初心者の方の学習や、忘れてしまった方の復習として、参考にしていただければ幸いです。

記載しているプログラムは、Ruby2.7.1を使って動作確認をしています。

Rubyのマルチスレッド処理
時間のかかる処理は非同期処理で効率よく実行しよう

Rubyのマルチスレッド処理

Rubyのマルチスレッド処理について、以下の内容で解説します。

マルチスレッドを扱うクラス
スレッドの作成と起動
スレッドからの戻り値の取得
簡易版スレッドプールの作成

マルチスレッドを扱うクラス

Rubyでスレッドを扱うには、標準ライブラリとして提供されている Threadクラスを使用します。

スレッドの作成と起動

スレッドを作成して起動するには、Threadクラスのnew、start、forkを使います。
newはinitializeを実行しますが、startとforkは実行しない違いがあります。

def threadA(name)
    puts "#{name} start."
    sleep(rand(0.0..3.0))
    puts "#{name} end."
end

puts "スレッド開始"
t1 = Thread.new {threadA("スレッドA-1")}
t2 = Thread.start {threadA("スレッドA-2")}
t3 = Thread.fork {threadA("スレッドA-3")}
t1.join
t2.join
t3.join
puts "スレッド終了"

上記のプログラムでは3種類の方法で作成したスレッドを起動しています。
スレッドで実行するメソッド threadA は3秒以下のランダム時間スリープする処理を行います。

joinは、処理が終了するまで待機します。
上記のプログラムの実行結果は以下となります。

スレッド開始
スレッドA-1 start.
スレッドA-2 start.
スレッドA-3 start.
スレッドA-2 end.
スレッドA-1 end.
スレッドA-3 end.
スレッド終了

3つのスレッドが同時に起動し、終了するまで待機していることが分かります。

スレッドからの戻り値の取得

スレッドが戻り値を返す場合は、valueで取得します。
valueはjoinと同じく処理が終了するまで待機します。

def threadB(name)
    puts "#{name} start."
    time = rand(0.0..3.0) 
    sleep(time)
    puts "#{name} end."
    return "#{name} #{time}"
end

puts "スレッド開始"
t4 = Thread.new {threadB("スレッドB-1")}
t5 = Thread.start {threadB("スレッドB-2")}
t6 = Thread.fork {threadB("スレッドB-3")}
puts t4.value
puts t5.value
puts t6.value
puts "スレッド終了"

上記のプログラムでは、スレッドで実行するメソッド methodB が戻り値を返します。
実行結果は以下のようになります。

スレッド開始
スレッドB-2 start.
スレッドB-3 start.
スレッドB-1 start.
スレッドB-1 end.
スレッドB-1 0.2343104981941182
スレッドB-3 end.
スレッドB-2 end.
スレッドB-2 2.3483202572798447
スレッドB-3 1.465095243124832
スレッド終了

スレッドが終了したものから戻り値を取得していることが分かります。

簡易版スレッドプールの作成

スレッドはいくつでも作って並列で処理させることが可能ですが、数が多すぎると逆に効率が悪くなることがあります。
スレッドが処理する数を制限する際に用いるのが、スレッドプールです。

スレッドプールは、スレッドの登録はいくつでも出来ますが、処理する最大数は決められた数までになります。
最大数は基本的に処理するマシンのCPUの数にすることが多いです。

Rubyでは、スレッドプールの機能は標準で用意されていないのですが、Thread::QueueまたはThead::SizedQueueを使って実装することで、簡易的にスレッドプールを作成することが可能です。

Thread::Queueを使ったスレッドプールの実装

queue = Thread::Queue.new
3.times do
    queue.push(:lock)
end

puts "スレッド開始"
Array.new(5) do |i|
    Thread.start do
        lock = queue.pop
        threadB("スレッドC-#{i+1}")
        queue.push(lock)
    end
end.each(&:join)
puts "スレッド終了"

上記のプログラムでは、最初にキューを作成し、3つの値（ロック）を入れています。
スレッドの作成・起動において、キューからロックを取得してスレッドを起動し、完了したらロックをキューに戻します。

5つのスレッドを作成していますが、キューからロックが取得できない場合は待機状態となり、ロックが取得できるとスレッドが起動します。
このようにすることで、最大3スレッドが同時に起動するスレッドプールのような形にすることができます。

上記のプログラムを実行すると、以下の結果となります。

スレッド開始
スレッドC-2 start.
スレッドC-3 start.
スレッドC-1 start.
スレッドC-1 end.
スレッドC-4 start.
スレッドC-2 end.
スレッドC-5 start.
スレッドC-5 end.
スレッドC-3 end.
スレッドC-4 end.
スレッド終了

上記の結果では、スレッド2、3、1が最初に起動し、スレッド4がスレッド1の終了まで待っています。
その後、スレッド2の終了後に、スレッド5が開始されていて、最大実行数が3になっていることが分かります。

Thread::SizedQueueを使ったスレッドプールの実装

SizedQueueは、キューのサイズを最初に決定します。
そのため、キューにロックを入れられる数をスレッドの最大実行数とすることができます。

queue = Thread::SizedQueue.new(3)

puts "スレッド開始"
Array.new(5) do |i|
    Thread.start do
        queue.push(:lock)
        threadB("スレッドD-#{i+1}")
        queue.pop
    end
end.each(&:join)
puts "スレッド終了"

今度はスレッドを開始する際に、キューにロックを入れ、終了したら取り出します。
キューにロックが入れられない場合は待機状態となり、他のスレッドが完了してキューが空いたらロックを入れてスレッドを開始します。

スレッド開始
スレッドD-2 start.
スレッドD-3 start.
スレッドD-4 start.
スレッドD-4 end.
スレッドD-1 start.
スレッドD-2 end.
スレッドD-5 start.
スレッドD-3 end.
スレッドD-1 end.
スレッドD-5 end.
スレッド終了

上記の結果では、スレッド2、3、4が最初に起動し、スレッド1がスレッド4の終了まで待ってます。
その後、スレッド2の終了後に、スレッド5が開始されていて、最大実行数が3になっていることが分かります。