Threading in Ruby mit einem Limit

Question

Ich habe eine Aufgabe, die ich durchführen muss, do_stuff(opts), die ~ 1s jedes dauert, auch während 1 - 10 von ihnen parallel ausgeführt werden. Ich muss am Ende ein Array der Ergebnisse für jede Operation sammeln.

Wenn ich 30 Dinge zu tun habe, wie würde ich effektiv Threading verwenden, um die do_stuff(opts) Operationen in Warteschlange, so dass nicht mehr als 10 gleichzeitig ausgeführt werden, aber das Array von Ergebnissen nicht gegeben/gedruckt/etc bis alle (30) Aufgaben wurden abgeschlossen?

Ich habe normalerweise zumindest etwas Code zu versuchen und zu illustrieren, was ich meine, aber mit Threading bin ich ein bisschen Verlust! Vielen Dank im Voraus

Answer 1

Sie benötigen this pattern
This question implementieren werden erläutert, wie das kann

Answer 2

auch in Ruby getan werden, einen Blick auf this Tutorial, wenn Sie zu Ruby-Threads neu sind.

Answer 3

Wenn Sie wirklich nach der Leistung sind, können Sie auch in jruby suchen.
Es nutzt tatsächliche OS Threads und nicht die grünen Fäden der andere Rubin Implementierungen

Answer 4

verwendet ich weiß nicht, wie es für eine komplexere Anwendung wird gut funktionieren, aber ich fand, so etwas wie dies für ein einfaches Einfädeln Szenario gut funktionieren mit Makrik.

thread_limit = 4 

threads = [] 
things_to_process.each do |thing| 
    until threads.map { |t| t.status }.count("run") < thread_limit do sleep 5 end 
    threads << Thread.new { the_task(thing) } 
end 
output = threads.map { |t| t.value } 

UNTIL Schleife um wartet, bis es weniger als die angegebene Anzahl von Threads erstellt läuft vor der Ausführung des Hauptthreads so dass der nächsten Thread weiterhin auf starten.

Die Ausgabevariable wird einem Array der von the_task zurückgegebenen Werte mit einer Reihenfolge zugewiesen, die dem Eingabearray things_to_process entspricht. Der Hauptthread wird blockiert, bis jeder erstellte Thread einen Wert zurückgibt.

Answer 5

Diese Lösung sammelt Ergebnisse im Array $ results. Es ermöglicht das Erstellen von 'thread_limit'-Threads und wartet darauf, dass sie abgeschlossen werden, bevor weitere erstellt werden.

$results = [] 

    def do_stuff(opts={}) 
    'done' 
    end 

    def thread_wait(threads) 
    threads.each{|t| t.join} 
    threads.each {|t| $results << t } 
    threads.delete_if {|t| t.status == false} 
    threads.delete_if {|t| t.status.nil? } 
    end 

    opts = {} 
    thread_limit = 20 
    threads = [] 
    records.each do |r| 
    thread_wait(threads) while threads.length >= thread_limit 
    t = Thread.new { do_stuff(opts) } 
    t.abort_on_exception = true 
    threads << t 
    end 
    # Ensure remaining threads complete 
    threads.each{|t| t.join} 

Answer 6

Ich benutze parals und paralsmap:

def parals(objects, n: 50) 
    objects.shuffle.each_slice(n).map do |g| 
    print '{' 
    threads = [] 
    g.map { |i| threads << Thread.new(i) { |i| yield(i) } } 
    threads.each(&:join) 
    print '}' 
    end 
end 

def paralsmap(objects, n: 50) 
    res = [] 

    objects.each_slice(n).map do |g| 
    print '{' 
    threads = [] 
    g.map { |i| threads << Thread.new(i, res) { |i| res << yield(i) } } 
    threads.each(&:join) 
    print '}' 
    end 

    res 
end 

z.B .:

parals((0..100).to_a) { |i| puts i } 
urls = parals((0..100).to_a) { |i| "https://google.com/?q=#{i}" } 

Sie die n Parameter die Anzahl der Threads begrenzen können.