Ruby - Threads and Dir [] arrays

Question

我已經構建了一個ruby腳本來創建文件夾，將文件移動到該新文件夾中，然後調用一個system()調用來觸發FFMPEG。現在我已將它轉換爲4個線程，以便我可以一次執行4個併發轉碼。

這裏是2個線程（減去文件夾結構創建和文件移動功能）的示例：

def transcode_process_1 

Dir["path/file/source/*.mxf"].each do |f| 

    random_folder = #code for random folder creation 

    file_move = #code to move .mxf file to random_folder for processing 

    system("FFMPEG -i #{random_folder} command 2> /path/file/#{random_filename}.txt") 

    sleep(2) 

end 
end 

def transcode_process_2 

sleep(3) 

Dir["path/file/source/*.mxf"].each do |f| 

    random_folder = #code for random folder creation 

    file_move = #code to move .mxf file to random_folder for processing 

    system("FFMPEG -i #{random_folder} command 2> /path/file/#{random_filename}.txt") 

    sleep(4) 

end 
end 

transcode_thread_1 = Thread.new{transcode_process_1()} 
transcode_thread_2 = Thread.new{transcode_process_2()} 

transcode_thread_1.join 
transcode_thread_2.join 

此遍歷迪爾「路徑/文件/源極/」並處理髮現的任何.mxf文件。我遇到的問題是，當兩個線程都在運行時，它們將相同的文件添加到數組中。這導致FFMPEG指出它找不到該文件（這是因爲另一個線程已經處理它並將其移至臨時文件夾）並創建多餘的文件夾和日誌文件，基本上只是使其變得凌亂。

我的問題是我將如何去確保transcode_thread_2不處理文件transcode_thread_1已添加到它的數組？有沒有辦法讓我可以檢查數組中的文件是否仍然存在？如果確實如此，那麼執行該過程，如果不移動到下一個文件？

感謝

Answer 1

有解決這個問題兩種方式，但我會考慮的第二種方法是優越。

1 - 檢查文件是否繼續

之前存在一個簡單的File.exist?(f)檢查添加到您的代碼：

def transcode_process_1 
    Dir["path/file/source/*.mxf"].each do |f| 
     next if !File.exist?(f) 
     ..carry on as normal.. 

2 - 使用隊列（該主題包的一部分）

你可以使用所有線程都可訪問的隊列。這簡化事情多種方式

1. 中央項目列表處理
2. 所有線程訪問，則可以生成儘可能多的線程，你可以處理
3. 刪除相同的代碼和那些的多個版本code smell睡眠值沿着線

東西：

require 'thread' 

total_threads = 2 # Or however cores you have  

transcode_queue = Queue.new 
transcode_queue << Dir["path/file/source/*.mxf"] 

def transcode_process(queue) 
    while !queue.empty? 
     file = queue.pop # Process this file 

     random_folder = '' #code for random folder creation 
     random_filename = '' # this was missing from your example 

     file_move = '' #code to move .mxf file to random_folder for processing 

     system("FFMPEG -i #{random_folder} command 2> /path/file/#{random_filename}.txt") 
    end 
end 

(1..total_threads).each { 
    Thread.new { transcode_process(transcode_queue) } 
} 

你並不需要sleep()，因爲不應該有任何競爭條件進入隊列，並且你已經爲每個新線程共享相同的代碼。

Answer 2

我沒有看到你所談論的數組，但這裏是你如何可以與一個普通蓄電池其跟蹤處理數據的排序了這一點：

acc = [] 

t1 = Thread.new do 
    a1 = [] 
    (1..10).each do |i| 
    unless acc.include?(i) 
     acc << i 
     a1 << i 
     p "a1 << #{i}" 
    end 
    end 
    p "a1 = #{a1}" 
end 

t2 = Thread.new do 
    a2 = [] 
    (1..10).each do |i| 
    unless acc.include?(i) 
     acc << i 
     a2 << i 
     p "a2 << #{i}" 
    end 
    end 
    p "a2 = #{a2}" 
end 

t1.join 
t2.join