Ruby 多線程

每個(gè)正在系統(tǒng)上運(yùn)行的程序都是一個(gè)進(jìn)程。每個(gè)進(jìn)程包含一到多個(gè)線程。

線程是程序中一個(gè)單一的順序控制流程，在單個(gè)程序中同時(shí)運(yùn)行多個(gè)線程完成不同的工作,稱為多線程。

Ruby 中我們可以通過 Thread 類來創(chuàng)建多線程，Ruby的線程是一個(gè)輕量級(jí)的，可以以高效的方式來實(shí)現(xiàn)并行的代碼。

創(chuàng)建 Ruby 線程

要啟動(dòng)一個(gè)新的線程，只需要調(diào)用 Thread.new 即可:

# 線程 #1 代碼部分
Thread.new {
  # 線程 #2 執(zhí)行代碼
}
# 線程 #1 執(zhí)行代碼

實(shí)例

以下實(shí)例展示了如何在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}"

以上代碼執(zhí)行結(jié)果為：

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、線程的創(chuàng)建可以使用Thread.new,同樣可以以同樣的語(yǔ)法使用Thread.start 或者Thread.fork這三個(gè)方法來創(chuàng)建線程。

2、創(chuàng)建線程后無需啟動(dòng)，線程會(huì)自動(dòng)執(zhí)行。

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

4、線程代碼塊中最后一個(gè)語(yǔ)句是線程的值，可以通過線程的方法來調(diào)用，如果線程執(zhí)行完畢，則返回線程值，否則不返回值直到線程執(zhí)行完畢。

5、Thread.current 方法返回表示當(dāng)前線程的對(duì)象。 Thread.main 方法返回主線程。

6、通過 Thread.Join 方法來執(zhí)行線程，這個(gè)方法會(huì)掛起主線程，直到當(dāng)前線程執(zhí)行完畢。

線程狀態(tài)

線程有5種狀態(tài)：

線程和異常

當(dāng)某線程發(fā)生異常，且沒有被rescue捕捉到時(shí)，該線程通常會(huì)被無警告地終止。但是，若有其它線程因?yàn)門hread#join的關(guān)系一直等待該線程的話，則等待的線程同樣會(huì)被引發(fā)相同的異常。

begin
  t = Thread.new do
    Thread.pass    # 主線程確實(shí)在等join
    raise "unhandled exception"
  end
  t.join
rescue
  p $!  # => "unhandled exception"
end

使用下列3個(gè)方法，就可以讓解釋器在某個(gè)線程因異常而終止時(shí)中斷運(yùn)行。

• 啟動(dòng)腳本時(shí)指定-d選項(xiàng)，并以調(diào)試模時(shí)運(yùn)行。
• 用Thread.abort_on_exception設(shè)置標(biāo)志。
• 使用Thread#abort_on_exception對(duì)指定的線程設(shè)定標(biāo)志。

當(dāng)使用上述3種方法之一后，整個(gè)解釋器就會(huì)被中斷。

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

線程同步控制

在Ruby中，提供三種實(shí)現(xiàn)同步的方式，分別是：

1. 通過Mutex類實(shí)現(xiàn)線程同步

2. 監(jiān)管數(shù)據(jù)交接的Queue類實(shí)現(xiàn)線程同步

3. 使用ConditionVariable實(shí)現(xiàn)同步控制

通過Mutex類實(shí)現(xiàn)線程同步

通過Mutex類實(shí)現(xiàn)線程同步控制，如果在多個(gè)線程鐘同時(shí)需要一個(gè)程序變量，可以將這個(gè)變量部分使用lock鎖定。 代碼如下：

#encoding:gbk
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

輸出結(jié)果如下：

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同步對(duì)某一個(gè)變量的訪問。

監(jiān)管數(shù)據(jù)交接的Queue類實(shí)現(xiàn)線程同步

Queue類就是表示一個(gè)支持線程的隊(duì)列，能夠同步對(duì)隊(duì)列末尾進(jìn)行訪問。不同的線程可以使用統(tǒng)一個(gè)對(duì)類，但是不用擔(dān)心這個(gè)隊(duì)列中的數(shù)據(jù)是否能夠同步，另外使用SizedQueue類能夠限制隊(duì)列的長(zhǎng)度

SizedQueue類能夠非常便捷的幫助我們開發(fā)線程同步的應(yīng)用程序，應(yīng)為只要加入到這個(gè)隊(duì)列中，就不用關(guān)心線程的同步問題。

經(jīng)典的生產(chǎn)者消費(fèi)者問題：

#!/usr/bin/ruby

require "thread"
puts "SizedQuee Test"

queue = Queue.new

producer = Thread.new do
     10.times do |i|
          sleep rand(i) # 讓線程睡眠一段時(shí)間
          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

線程變量

線程可以有其私有變量，線程的私有變量在線程創(chuàng)建的時(shí)候?qū)懭刖€程?？梢员痪€程范圍內(nèi)使用，但是不能被線程外部進(jìn)行共享。

但是有時(shí)候，線程的局部變量需要?jiǎng)e別的線程或者主線程訪問怎么辦？ruby當(dāng)中提供了允許通過名字來創(chuàng)建線程變量，類似的把線程看做hash式的散列表。通過[]=寫入并通過[]讀出數(shù)據(jù)。我們來看一下下面的代碼：

#!/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}"

以上代碼運(yùn)行輸出結(jié)果為：

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

主線程等待子線程執(zhí)行完成，然后分別輸出每個(gè)值。 。

線程優(yōu)先級(jí)

線程的優(yōu)先級(jí)是影響線程的調(diào)度的主要因素。其他因素包括占用CPU的執(zhí)行時(shí)間長(zhǎng)短，線程分組調(diào)度等等。

可以使用 Thread.priority 方法得到線程的優(yōu)先級(jí)和使用 Thread.priority= 方法來調(diào)整線程的優(yōu)先級(jí)。

線程的優(yōu)先級(jí)默認(rèn)為 0 。 優(yōu)先級(jí)較高的執(zhí)行的要快。

一個(gè) Thread 可以訪問自己作用域內(nèi)的所有數(shù)據(jù)，但如果有需要在某個(gè)線程內(nèi)訪問其他線程的數(shù)據(jù)應(yīng)該怎么做呢？ Thread 類提供了線程數(shù)據(jù)互相訪問的方法，你可以簡(jiǎn)單的把一個(gè)線程作為一個(gè) Hash 表，可以在任何線程內(nèi)使用 []= 寫入數(shù)據(jù)，使用 [] 讀出數(shù)據(jù)。

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"]}" }

可以看到，把線程作為一個(gè) Hash 表，使用 [] 和 []= 方法，我們實(shí)現(xiàn)了線程之間的數(shù)據(jù)共享。

線程互斥

Mutex(Mutal Exclusion = 互斥鎖)是一種用于多線程編程中，防止兩條線程同時(shí)對(duì)同一公共資源（比如全局變量）進(jìn)行讀寫的機(jī)制。

不使用Mutax的實(shí)例

#!/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}"

以上實(shí)例運(yùn)行輸出結(jié)果為：

count1 :  9712487
count2 :  12501239
difference : 0

使用Mutax的實(shí)例

#!/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}"

以上實(shí)例運(yùn)行輸出結(jié)果為：

count1 :  1336406
count2 :  1336406
difference : 0

死鎖

兩個(gè)以上的運(yùn)算單元，雙方都在等待對(duì)方停止運(yùn)行，以獲取系統(tǒng)資源，但是沒有一方提前退出時(shí)，這種狀況，就稱為死鎖。

例如，一個(gè)進(jìn)程 p1占用了顯示器，同時(shí)又必須使用打印機(jī)，而打印機(jī)被進(jìn)程p2占用，p2又必須使用顯示器，這樣就形成了死鎖。

當(dāng)我們?cè)谑褂?Mutex 對(duì)象時(shí)需要注意線程死鎖。

實(shí)例

#!/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

以上實(shí)例輸出結(jié)果為：

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（線程） 類方法如下：

線程實(shí)例化方法

以下實(shí)例調(diào)用了線程實(shí)例化方法 join：

#!/usr/bin/ruby

thr = Thread.new do   # 實(shí)例化
   puts "In second thread"
   raise "Raise exception"
end
thr.join   # 調(diào)用實(shí)例化方法 join

以下是完整實(shí)例化方法列表：