Gwintowanie w Ruby z limitem

Question

Mam zadanie, które muszę wykonać, do_stuff(opts), które zajmie ~ 1s każdy, nawet gdy 1 - 10 z nich pracuje równolegle. Muszę zebrać tablicę wyników dla każdej operacji na końcu.

Jeśli mam 30 rzeczy do zrobienia, w jaki sposób mogę efektywnie używać wątków do kolejkowania operacji do_stuff(opts), więc nie więcej niż 10 działa równolegle, ale tablica wyników nie jest podawana/drukowana/etc aż do wszystkich (30) zadania zostały zakończone?

Zwykle mam przynajmniej trochę kodu, aby spróbować i zilustrować, co mam na myśli, ale z gwintowaniem jestem trochę zagubiony! Dzięki z góry

Answer 1

Trzeba wdrożyć this pattern
This question omawia, w jaki sposób można zrobić w Ruby

Answer 2

Również spojrzeć na this tutorialu jeśli jesteś nowy w Ruby wątkach.

Answer 3

Jeśli naprawdę zależy Ci na wydajności, możesz również zajrzeć do jruby.
Wykorzystuje rzeczywiste wątki OS, a nie zielone wątki inne implementacje Ruby korzystają

Answer 4

Nie wiem, jak dobrze będzie pracować dla bardziej złożonych aplikacji, ale znalazłem coś takiego działa ładnie prostego scenariusza gwintowania z macruby.

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 } 

czeka aż pętla wokół dopóki istnieją mniej niż określoną liczbę utworzonych wątków uruchomionych przed zezwoleniem na wykonanie głównego wątku, aby kontynuować, aby rozpocząć kolejny wątek.

do zmiennej wyjściowej zostanie przypisana tablica wartości zwróconych przez_taskę z zamówieniem odpowiadającym wejściowej tablicy things_to_process. Główny wątek będzie blokowany, dopóki każdy utworzony wątek nie zwróci wartości.

Answer 5

To rozwiązanie gromadzi wyniki w tablicy wyników $. Umożliwia tworzenie wątków "thread_limit", a następnie czeka na ich ukończenie przed utworzeniem kolejnych.

$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

używam parals i 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 

np .:

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

Można użyć parametru n ograniczyć liczbę wątków.