Latest web development tutorials

Ruby 多線程

每個正在系統上運行的程序都是一個進程。 每個進程包含一到多個線程。

線程是程序中一個單一的順序控制流程,在單個程序中同時運行多個線程完成不同的工作,稱為多線程。

Ruby 中我們可以通過Thread 類來創建多線程,Ruby的線程是一個輕量級的,可以以高效的方式來實現並行的代碼。


創建Ruby 線程

要啟動一個新的線程,只需要調用Thread.new 即可:

# 线程 #1 代码部分
Thread.new {
  # 线程 #2 执行代码
}
# 线程 #1 执行代码

實例

以下實例展示瞭如何在Ruby程序中使用多線程:

#!/usr/bin/ruby

def func1
   i=0
   while i<=2
      puts "func1 at: #{Time.now}"
      sleep(2)
      i=i+1
   end
end

def func2
   j=0
   while j<=2
      puts "func2 at: #{Time.now}"
      sleep(1)
      j=j+1
   end
end

puts "Started At #{Time.now}"
t1=Thread.new{func1()}
t2=Thread.new{func2()}
t1.join
t2.join
puts "End at #{Time.now}"

以上代碼執行結果為:

Started At Wed May 14 08:21:54 -0700 2014
func1 at: Wed May 14 08:21:54 -0700 2014
func2 at: Wed May 14 08:21:54 -0700 2014
func2 at: Wed May 14 08:21:55 -0700 2014
func1 at: Wed May 14 08:21:56 -0700 2014
func2 at: Wed May 14 08:21:56 -0700 2014
func1 at: Wed May 14 08:21:58 -0700 2014
End at Wed May 14 08:22:00 -0700 2014

線程生命週期

1、線程的創建可以使用Thread.new,同樣可以以同樣的語法使用Thread.start 或者Thread.fork這三個方法來創建線程。

2、創建線程後無需啟動,線程會自動執行。

3、Thread 類定義了一些方法來操控線程。 線程執行Thread.new中的代碼塊。

4、線程代碼塊中最後一個語句是線程的值,可以通過線程的方法來調用,如果線程執行完畢,則返迴線程值,否則不返回值直到線程執行完畢。

5、Thread.current 方法返回表示當前線程的對象。 Thread.main 方法返回主線程。

6、通過Thread.Join 方法來執行線程,這個方法會掛起主線程,直到當前線程執行完畢。


線程狀態

線程有5種狀態:

線程狀態 返回值
可執行 run
睡眠 Sleeping
退出 aborting
正常終止 false
發生異常終止 nil

線程和異常

當某線程發生異常,且沒有被rescue捕捉到時,該線程通常會被無警告地終止。 但是,若有其它線程因為Thread#join的關係一直等待該線程的話,則等待的線程同樣會被引發相同的異常。

begin
  t = Thread.new do
    Thread.pass    # 主线程确实在等join
    raise "unhandled exception"
  end
  t.join
rescue
  p $!  # => "unhandled exception"
end

使用下列3個方法,就可以讓解釋器在某個線程因異常而終止時中斷運行。

  • 啟動腳本時指定-d選項,並以調試模時運行。
  • Thread.abort_on_exception設置標誌。
  • 使用Thread#abort_on_exception對指定的線程設定標誌。

當使用上述3種方法之一後,整個解釋器就會被中斷。

t = Thread.new { ... }
t.abort_on_exception = true

線程同步控制

在Ruby中,提供三種實現同步的方式,分別是:

1. 通過Mutex類實現線程同步

2. 監管數據交接的Queue類實現線程同步

3. 使用ConditionVariable實現同步控制

通過Mutex類實現線程同步

通過Mutex類實現線程同步控制,如果在多個線程鐘同時需要一個程序變量,可以將這個變量部分使用lock鎖定。 代碼如下:

#!/usr/bin/ruby

require "thread"
puts "Synchronize Thread"

@num=200
@mutex=Mutex.new

def buyTicket(num)
     @mutex.lock
          if @num>=num
               @num=@num-num
               puts "you have successfully bought #{num} tickets"
          else
               puts "sorry,no enough tickets"
          end
     @mutex.unlock
end

ticket1=Thread.new 10 do
     10.times do |value|
     ticketNum=15
     buyTicket(ticketNum)
     sleep 0.01
     end
end

ticket2=Thread.new 10 do
     10.times do |value|
     ticketNum=20
     buyTicket(ticketNum)
     sleep 0.01
     end
end

sleep 1
ticket1.join
ticket2.join

以上代碼執行結果為:

Synchronize Thread
you have successfully bought 15 tickets
you have successfully bought 20 tickets
you have successfully bought 15 tickets
you have successfully bought 20 tickets
you have successfully bought 15 tickets
you have successfully bought 20 tickets
you have successfully bought 15 tickets
you have successfully bought 20 tickets
you have successfully bought 15 tickets
you have successfully bought 20 tickets
you have successfully bought 15 tickets
sorry,no enough tickets
sorry,no enough tickets
sorry,no enough tickets
sorry,no enough tickets
sorry,no enough tickets
sorry,no enough tickets
sorry,no enough tickets
sorry,no enough tickets
sorry,no enough tickets

除了使用lock鎖定變量,還可以使用try_lock鎖定變量,還可以使用Mutex.synchronize同步對某一個變量的訪問。

監管數據交接的Queue類實現線程同步

Queue類就是表示一個支持線程的隊列,能夠同步對隊列末尾進行訪問。 不同的線程可以使用統一個對類,但是不用擔心這個隊列中的數據是否能夠同步,另外使用SizedQueue類能夠限制隊列的長度

SizedQueue類能夠非常便捷的幫助我們開發線程同步的應用程序,應為只要加入到這個隊列中,就不用關心線程的同步問題。

經典的生產者消費者問題:

#!/usr/bin/ruby

require "thread"
puts "SizedQuee Test"

queue = Queue.new

producer = Thread.new do
     10.times do |i|
          sleep rand(i) # 让线程睡眠一段时间
          queue << i
          puts "#{i} produced"
     end
end

consumer = Thread.new do
     10.times do |i|
          value = queue.pop
          sleep rand(i/2)
          puts "consumed #{value}"
     end
end

consumer.join

程序的輸出:

SizedQuee Test
0 produced
1 produced
consumed 0
2 produced
consumed 1
consumed 2
3 produced
consumed 34 produced

consumed 4
5 produced
consumed 5
6 produced
consumed 6
7 produced
consumed 7
8 produced
9 produced
consumed 8
consumed 9

線程變量

線程可以有其私有變量,線程的私有變量在線程創建的時候寫入線程。 可以被線程範圍內使用,但是不能被線程外部進行共享。

但是有時候,線程的局部變量需要別別的線程或者主線程訪問怎麼辦? ruby當中提供了允許通過名字來創建線程變量,類似的把線程看做hash式的散列表。 通過[]=寫入並通過[]讀出數據。 我們來看一下下面的代碼:

#!/usr/bin/ruby

count = 0
arr = []

10.times do |i|
   arr[i] = Thread.new {
      sleep(rand(0)/10.0)
      Thread.current["mycount"] = count
      count += 1
   }
end

arr.each {|t| t.join; print t["mycount"], ", " }
puts "count = #{count}"

以上代碼運行輸出結果為:

8, 0, 3, 7, 2, 1, 6, 5, 4, 9, count = 10

主線程等待子線程執行完成,然後分別輸出每個值。 。


線程優先級

線程的優先級是影響線程的調度的主要因素。 其他因素包括佔用CPU的執行時間長短,線程分組調度等等。

可以使用Thread.priority 方法得到線程的優先級和使用Thread.priority= 方法來調整線程的優先級。

線程的優先級默認為0 。 優先級較高的執行的要快。

一個Thread 可以訪問自己作用域內的所有數據,但如果有需要在某個線程內訪問其他線程的數據應該怎麼做呢? Thread 類提供了線程數據互相訪問的方法,你可以簡單的把一個線程作為一個Hash 表,可以在任何線程內使用[]= 寫入數據,使用[] 讀出數據。

athr = Thread.new { Thread.current["name"] = "Thread A"; Thread.stop }
bthr = Thread.new { Thread.current["name"] = "Thread B"; Thread.stop }
cthr = Thread.new { Thread.current["name"] = "Thread C"; Thread.stop }
Thread.list.each {|x| puts "#{x.inspect}: #{x["name"]}" }

可以看到,把線程作為一個Hash 表,使用[] 和[]= 方法,我們實現了線程之間的數據共享。


線程互斥

Mutex(Mutal Exclusion = 互斥鎖)是一種用於多線程編程中,防止兩條線程同時對同一公共資源(比如全局變量)進行讀寫的機制。

不使用Mutax的實例

#!/usr/bin/ruby
require 'thread'

count1 = count2 = 0
difference = 0
counter = Thread.new do
   loop do
      count1 += 1
      count2 += 1
   end
end
spy = Thread.new do
   loop do
      difference += (count1 - count2).abs
   end
end
sleep 1
puts "count1 :  #{count1}"
puts "count2 :  #{count2}"
puts "difference : #{difference}"

以上實例運行輸出結果為:

count1 :  9712487
count2 :  12501239
difference : 0

使用Mutax的實例

#!/usr/bin/ruby
require 'thread'
mutex = Mutex.new

count1 = count2 = 0
difference = 0
counter = Thread.new do
   loop do
      mutex.synchronize do
         count1 += 1
         count2 += 1
      end
    end
end
spy = Thread.new do
   loop do
       mutex.synchronize do
          difference += (count1 - count2).abs
       end
   end
end
sleep 1
mutex.lock
puts "count1 :  #{count1}"
puts "count2 :  #{count2}"
puts "difference : #{difference}"

以上實例運行輸出結果為:

count1 :  1336406
count2 :  1336406
difference : 0

死鎖

兩個以上的運算單元,雙方都在等待對方停止運行,以獲取系統資源,但是沒有一方提前退出時,這種狀況,就稱為死鎖。

例如,一個進程p1佔用了顯示器,同時又必須使用打印機,而打印機被進程p2佔用,p2又必須使用顯示器,這樣就形成了死鎖。

當我們在使用Mutex 對象時需要注意線程死鎖。

實例

#!/usr/bin/ruby
require 'thread'
mutex = Mutex.new

cv = ConditionVariable.new
a = Thread.new {
   mutex.synchronize {
      puts "A: I have critical section, but will wait for cv"
      cv.wait(mutex)
      puts "A: I have critical section again! I rule!"
   }
}

puts "(Later, back at the ranch...)"

b = Thread.new {
   mutex.synchronize {
      puts "B: Now I am critical, but am done with cv"
      cv.signal
      puts "B: I am still critical, finishing up"
   }
}
a.join
b.join

以上實例輸出結果為:

A: I have critical section, but will wait for cv
(Later, back at the ranch...)
B: Now I am critical, but am done with cv
B: I am still critical, finishing up
A: I have critical section again! I rule!

線程類方法

完整的Thread(線程) 類方法如下:

序號 方法描述
1 Thread.abort_on_exception
若其值為真的話,一旦某線程因異常而終止時,整個解釋器就會被中斷。 它的默認值是假,​​也就是說,在通常情況下,若某線程發生異常且該異常未被Thread#join等檢測到時,該線程會被無警告地終止。
2 Thread.abort_on_exception=
如果設置為true ,一旦某線程因異常而終止時,整個解釋器就會被中斷。 返回新的狀態
3 Thread.critical
返回布爾值。
4 Thread.critical=
當其值為true時,將不會進行線程切換。 若當前線程掛起(stop)或有信號(signal)干預時,其值將自動變為false。
5 Thread.current
返回當前運行中的線程(當前線程)。
6 Thread.exit
終止當前線程的運行。 返回當前線程。 若當前線程是唯一的一個線程時,將使用exit(0)來終止它的運行。
7 Thread.fork { block }
與Thread.new 一樣生成線程。
8 Thread.kill( aThread )
終止線程的運行.
9 Thread.list
返回處於運行狀態或掛起狀態的活線程的數組。
10 Thread.main
返回主線程。
11 Thread.new( [ arg ]* ) {| args | block }
生成線程,並開始執行。 數會被原封不動地傳遞給塊. 這就可以在啟動線程的同時,將值傳遞給該線程所固有的局部變量。
12 Thread.pass
將運行權交給其他線程. 它不會改變運行中的線程的狀態,而是將控制權交給其他可運行的線程(顯式的線程調度)。
13 Thread.start( [ args ]* ) {| args | block }
生成線程,並開始執行。 數會被原封不動地傳遞給塊. 這就可以在啟動線程的同時,將值傳遞給該線程所固有的局部變量。
14 Thread.stop
將當前線程掛起,直到其他線程使用run方法再次喚醒該線程。

線程實例化方法

以下實例調用了線程實例化方法join:

#!/usr/bin/ruby

thr = Thread.new do   # 实例化
   puts "In second thread"
   raise "Raise exception"
end
thr.join   # 调用实例化方法 join

以下是完整實例化方法列表:

序號 方法描述
1 thr[ name ]
取出線程內與name相對應的固有數據。 name可以是字符串或符號。 若沒有與name相對應的數據時, 返回nil。
2 thr[ name ] =
設置線程內name相對應的固有數據的值, name可以是字符串或符號。 若設為nil時, 將刪除該線程內對應數據。
3 thr.abort_on_exception
返回布爾值。
4 thr.abort_on_exception=
若其值為true的話,一旦某線程因異常而終止時,整個解釋器就會被中斷。
5 thr.alive?
若線程是"活"的,就返回true。
6 thr.exit
終止線程的運行。 返回self。
7 thr.join
掛起當前線程,直到self線程終止運行為止. 若self因異常而終止時, 將會當前線程引發同樣的異常。
8 thr.key?
若與name相對應的線程固有數據已經被定義的話,就返回true
9 thr.kill
類似於Thread.exit
10 thr.priority
返迴線程的優先度. 優先度的默認值為0. 該值越大則優先度越高.
11 thr.priority=
設定線程的優先度. 也可以將其設定為負數.
12 thr.raise( anException )
在該線程內強行引發異常.
13 thr.run
重新啟動被掛起(stop)的線程. 與wakeup不同的是,它將立即進行線程的切換. 若對死進程使用該方法時, 將引發ThreadError異常.
14 thr.safe_level
返回self 的安全等級. 當前線程的safe_level與$SAFE相同.
15 thr.status
使用字符串"run"、"sleep"或"aborting" 來表示活線程的狀態. 若某線程是正常終止的話,就返回false. 若因異常而終止的話,就返回nil。
16 thr.stop?
若線程處於終止狀態(dead)或被掛起(stop)時,返回true.
17 thr.value
一直等到self線程終止運行(等同於join)後,返回該線程的塊的返回值. 若在線程的運行過程中發生了異常, 就會再次引發該異常.
18 thr.wakeup
把被掛起(stop)的線程的狀態改為可執行狀態(run), 若對死線程執行該方法時,將會引發ThreadError異常。