Thread是Ruby的線程庫，Thread庫已經(jīng)內(nèi)置在Ruby中，但如果想要使用線程安全的Queue、Mutex以及條件變量等，則需要手動require 'thread'。

主線程main

默認(rèn)情況下，每個Ruby進(jìn)程都具備一個主線程main，如果沒有創(chuàng)建新的線程，所有的代碼都將在這個主線程分支中執(zhí)行。

使用Thread.main()類方法可獲取當(dāng)前線程組的主線程，使用Thread.current()可以獲取當(dāng)前正在執(zhí)行的線程分支。使用Thread.list()可獲取當(dāng)前進(jìn)程組中所有存活的線程。

p Thread.main
p Thread.current
p Thread.main == Thread.current
=begin
#<Thread:0x0000000001d9ae58 run>
#<Thread:0x0000000001d9ae58 run>
true
=end

可見，線程其實是一個Thread類的實例對象。

創(chuàng)建Ruby線程

使用Thread庫的new()、start()、fork()可創(chuàng)建線程，它們幾乎等價，且后兩者是別名關(guān)系。

創(chuàng)建線程時需傳遞一個代碼塊或Proc對象參數(shù)， 它們是要執(zhí)行的任務(wù),它們將在新的線程分支中執(zhí)行。如果需要,可以為代碼塊或Proc對象傳遞參數(shù)。

arr=[]
a,b,C=1,2,3
Thread.new(a,b,c) { |d,e,f| arr << d << e << f }
sleep 1
p arr   #=> [1,2,3]

如果主線程先執(zhí)行完成，主線程將直接退出，主線程的退出將會終止進(jìn)程，使得其它線程也會退出。

Thread.new {puts "hello"}
puts "world"

上述代碼幾乎總是會輸出world，然后退出，主線程的退出使得子線程不會輸出"hello"。之所以總是會輸出world而不是輸出hello，這和Ruby的線程調(diào)度有關(guān)，在后面的文章中會詳細(xì)解釋Ruby中的線程調(diào)度。

join()和value()等待線程

如果想要等待某個線程先執(zhí)行完成，可使用t.join()，如果線程t尚未退出，則join()會阻塞。可以在任意線程中調(diào)用t.join()，誰調(diào)用誰等待。

t = Thread.new { puts "I am Child" }
t.join  # 等待子線程執(zhí)行完成
puts "I am Parent"

還可以將多個線程對象放進(jìn)數(shù)組，然后執(zhí)行遍歷join，另一種常見的做法是使用map{}.each(&:join)的方式：

threads = []
3.times do |i|
  # 將多個線程加入到數(shù)組中
  threads << Thread.new { puts "Thread #{i}" }
end

# 在main線程中join每個線程，
# 因此只有3個線程全都完成后，main線程才會繼續(xù)，即退出
threads.each(&:join)
=begin
Thread 1
Thread 0
Thread 2
=end

# 另一種常見方式
3.times.map {|i| Thread.new { puts "Thread #{i}" } }.each(&:join)
Array.new(3) {|i| Thread.new { puts "Thread #{i}" } }.each(&:join)

t.value()和t.join()類似，不同之處在于t.value()在內(nèi)部調(diào)用t.join()等待線程t之后，還會在等待成功時取得該線程的返回值。

a = Thread.new { 2 + 2 }
p a.value   #=> 4

注意，對于Ruby來說，無論是否執(zhí)行join()操作，任務(wù)執(zhí)行完成的線程都會馬上被操作系統(tǒng)回收(從OS線程表中刪除)，但被回收的線程仍然能夠使用value()方法來獲取被回收線程的返回值。之所以會這樣，我個人猜想，也許是因為Ruby內(nèi)部已經(jīng)幫我們執(zhí)行了join操作并將線程返回值保存在Ruby內(nèi)部，這樣對于用戶來說就更加安全，而且用戶執(zhí)行join()或value()操作，可能是在等待Ruby內(nèi)部的這個值的出現(xiàn)。

線程的異常處理

默認(rèn)情況下，當(dāng)某個非main線程中拋出異常后，該線程將因異常而終止，但是它的終止不會影響其它線程。

t = Thread.new {raise "hello"}    # 拋出異常
sleep 1    # 仍然睡眠1秒后退出

如果使用了t.join()或t.value()去等待拋出異常的線程t，異常將會傳播給調(diào)用這兩個方法的線程。例如主線程調(diào)用t.join，如果t會拋出一次異常，那么主線程在等待過程中還會拋出一次異常。

t = Thread.new {raise "hello"}    # 拋出異常
t.join()    # 子線程拋異常后，main線程也拋異常

如果想要讓任意線程出現(xiàn)異常時終止整個程序，可設(shè)置類方法Thread.abort_on_exception為true，它會在任意子線程拋出異常后自動傳播給main線程，從而終止進(jìn)程：

Thread.abort_on_exception = true
Thread.new { raise "Error" }
sleep 1   # 不會睡眠完1秒，而是子線程異常后立即異常退出

如果想要讓某個特定的線程出現(xiàn)異常時終止整個程序，可設(shè)置同名的實例方法t.abort_on_exception為true，只有t線程異常時才會終止程序。

t1 = Thread.new { raise "Error from t1" }
t1.abort_on_exception = true
sleep 1

另外，線程實例方法t.raise()可以直接在線程t拋出異常。

需注意，Ruby線程有一個巨大的缺點：無論是raise拋出異常還是各種終止(比如kill、exit)，都不會執(zhí)行ensure子句。

線程的狀態(tài)和生命周期

Ruby中的線程具有5種狀態(tài)，可通過t.status()查看，該方法有5種對應(yīng)的返回值：

- run: 線程正在運行(running)或可運行(runnable)  
- sleep: 線程處于睡眠態(tài)，比如阻塞(如sleep,mutex,io block)  
- false: 線程正常退出后的狀態(tài)，包括執(zhí)行完流程、手動退出(t.exit)、信號終止(t.kill)  
- nil: 線程因拋出異常(比如raise)而退出的狀態(tài)  
- aborting: 線程被完全kill之前的過渡狀態(tài)，不考慮這種狀態(tài)的存在

另外，還有兩種統(tǒng)稱狀態(tài)：

• alive：存活的線程，等價于run + sleep
• stop：已停止的線程，等價于sleep + dead(false+nil)

可分別使用alive?()和stop?()來判斷線程是否屬于這兩種統(tǒng)稱狀態(tài)。

此外：

Kernel.sleep：讓當(dāng)前線程睡眠指定時長，無參數(shù)則永久睡眠，線程將進(jìn)入睡眠隊列
Thread.stop：讓當(dāng)前線程睡眠，進(jìn)入睡眠隊列，等價于無參數(shù)的sleep  
Thread.pass：轉(zhuǎn)讓CPU，當(dāng)前線程進(jìn)入就緒隊列而不是睡眠隊列  
t.run：喚醒線程t使其進(jìn)入就緒隊列，同時讓當(dāng)前線程放棄CPU，調(diào)度程序?qū)⒅匦抡{(diào)度  
t.wakeup：喚醒線程t使其進(jìn)入就緒隊列，但不會讓當(dāng)前線程放棄CPU，調(diào)度程序?qū)⒉粫⒓粗匦抡{(diào)度  

Thread.kill：終止指定線程，它將不再被調(diào)度  
Thread.exit：終止當(dāng)前線程，它將不再被調(diào)度  
t.exit,t.kill,t.terminate：終止線程t，t將不再被調(diào)度

幾個注意事項：

• 這里5個終止線程的方式效果上是完全等價的，三個實例方法是別名關(guān)系，而兩個類方法的內(nèi)部也都是調(diào)用線程對象的kill
• 最好要不加區(qū)分地看待run和wakeup
• 對于Thread.pass，除了知道它轉(zhuǎn)讓CPU的行為是確定的，不要對它假設(shè)任何額外的行為，比如不要認(rèn)為出讓CPU后一定會調(diào)度到其它Ruby線程，很有可能會在調(diào)度其它一些非Ruby線程后再次先調(diào)度到本線程而非其它Ruby線程
• 需注意，無論是raise拋出異常還是各種終止(比如kill、exit)，都不會執(zhí)行ensure子句

線程私有變量和局部變量

Ruby進(jìn)程內(nèi)的所有線程共享進(jìn)程的虛擬地址空間，所以共享了一些數(shù)據(jù)。

但線程是語句塊或者Proc對象，所以語句塊內(nèi)部創(chuàng)建的變量是在當(dāng)前線程棧內(nèi)部的，是每個線程私有的變量。

# 主線程中的變量
a = 1

# 子線程
t1 = Thread.new(3) do |x|
  a += 1
  b=3
  x=4
end

# 主線程
t1.join
p a   # 2
#p b  # 報錯，b不存在
#p x  # 報錯，x不存在

Ruby為線程提供了局部變量共享的概念，每個線程對象都可以有自己的局部數(shù)據(jù)空間(即線程本地變量)，線程對象的局部空間互不影響，比如兩個線程中同時進(jìn)行正則匹配，兩個線程的$~是不一樣且互不影響的。

線程對象t的局部數(shù)據(jù)空間是t[key]=value，即一個名為t的hash結(jié)構(gòu)，因為對象t是可以共享的，所以它的局部空間也是共享的。

t1 = Thread.new do
  t = Thread.current
  t[:name] = "junmajinlong"
  t[:age] = 23
end

t1.join

p t1.keys          # [:name, :age]
p t1.key? :gender  # false
p t1[:name]        # "junmajinlong"
t1[:age] = 24  
p t1[:age]         # 24

所以，有這么幾個方法：

t[key]
t[key]=
t.keys
t.key?

此外還有一個fetch()方法，類似于Hash的fetch()，默認(rèn)情況下訪問不存在的key會異常，可指定默認(rèn)值或通過語句塊返回默認(rèn)值。

嚴(yán)格來說，從Ruby 1.9出現(xiàn)Fiber之后，t[]不再是線程本地變量(thread-local)，而是纖程(Fiber)本地變量(fiber-local)。但也支持使用線程本地變量：

t.thread_variables
t.thread_variable?
t.thread_variable_get
t.thread_variable_set

線程組

默認(rèn)情況下，所有線程都在默認(rèn)的線程組中，這個默認(rèn)線程組是Ruby程序啟動時創(chuàng)建的?？墒褂?code>ThreadGroup::Default獲取默認(rèn)線程組。

t1 = Thread.new do
  Thread.stop
end

p t1.group
p Thread.current.group
p ThreadGroup::Default
=begin
#<ThreadGroup:0x00000000019bcb60>
#<ThreadGroup:0x00000000019bcb60>
#<ThreadGroup:0x00000000019bcb60>
=end

• 使用ThreadGroup.new可創(chuàng)建一個自定義的線程組
• 使用tg.add(t)可將線程t加入線程組tg，這將會從原來的線程組移除t再加入新組tg
• 使用tg.list可列出線程組tg中的所有線程
• 使用t.group可獲取線程t所屬的線程組
• 子線程會繼承父線程的線程組，即子線程也會加入父線程所在的線程組

tg = ThreadGroup.new
t1 = Thread.new { Thread.stop }
t2 = Thread.new { Thread.stop }
tg.add t1
tg.add t2
pp tg.list
pp t1.group
=begin
[#<Thread:0x000000000196c480 a.rb:4 sleep_forever>,
 #<Thread:0x000000000196c3b8 a.rb:5 sleep_forever>]
#<ThreadGroup:0x000000000196c520>
=end

線程組還有一個功能：可使用tg.enclose封閉線程組tg，封閉后的線程組將不允許內(nèi)部線程移出加入其它組，也不允許外界線程加入該組，只允許在該組中創(chuàng)建新線程。使用tg.enclosed?測試線程組tg是否已封閉。

其實，使用線程組可以將多個線程分類統(tǒng)一管理，線程組本質(zhì)是一個線程數(shù)組加一些額外屬性。比如，可以為線程組定義一些額外的針對線程組中所有線程的功能：wakeup組中的所有線程、join所有線程、kill所有線程。

class ThreadGroup
  def wakeup
    list.each(&:wakeup)
  end
  def join
    list.each { |th| th.join if th != Thread.current }
  end
  def kill
    list.each(&:kill)
  end
end

更多關(guān)于Ruby多線程知識請查看下面的相關(guān)鏈接

最新評論