Threading and Looping in Ruby

Question

Я ищу способ, чтобы мои потоки могли проходить через массив адресов электронной почты, не наступая друг на друга и не меняя переменные (я не могу использовать мьютексы). Я нашел некоторую информацию об использовании «локальных переменных потока», но не может показаться, что это работает. Ниже приведен пример моей проблемы (это всего лишь небольшой патрон кода).

(1..(threads).map { |thread_count| 

    Thread.new do 

    (1..(messages).each do |message_count| 

    email = recipients_array[recipient_count].join(", ") 

    if (recipient_count != (recipients_array.length - 1)) 
     recipient_count += 1 
    else 
     recipient_count = 0 
    end 

Я застрял на этом некоторое время. Я пишу скрипт, который использует многопоточность в JRuby для отправки электронной почты. Я рассказываю сценарию, сколько потоков я хочу отправить и сколько сообщений в потоке я собираюсь отправить. Я передаю текстовый файл адресов получателей, которые загружаются в массив. Затем я хочу, чтобы перебирать массив таким образом, чтобы:

• Тема 1, сообщение 1 будет идти по электронной почте 1
• тему 2, сообщение 1 будет идти по электронной почте 1
• Тема 1, сообщение 2 будет идти по электронной почте 2
• тему 2, сообщение 2 будет идти по электронной почте 2

и так далее ... Это начинается хорошо, но если я устанавливаю, чтобы сделать 5 х 5 нитей сообщения:

• Темы с 1 по 5, 1 сообщение будет идти по электронной почте 1
• тему 1, 2 Сообщение будет идти по электронной почте 6

, потому что все они доступ к recipient_count переменной и приращением ее +1.

Ищете рекомендации относительно того, как лучше настроить это.

Answer 1

Обычно я использую рубин многопоточность таким образом:

require 'thread' 

count = 4 
result = [] 

mutex = Mutex.new 
queue = Queue.new 

# fill the queue 
(0..100).each do |i| 
    queue << i 
end 

(0..count).map do 
    begin 
    loop do 
     item = queue.pop(true) 
     item = do_something_with_it(item) 

     mutex.synchronize do 
     result << item 
     end 
    end 
    rescue ThreadError 
    Thread.exit 
    end 
end.each(&:join) 

# process results 

Answer 2

Вы сказали, что этот сценарий вызывается с threads, messages и recipient_array в качестве аргументов. Я не уверен, какую форму занимают отдельные записи в recipients_array или что такое Array#join, и почему вы сбросили recipient_count на 0, когда он достиг последнего индекса recipients_array. Я предполагаю, что есть недостающий код. Но, как насчет этого.

emails_handled = [] 
(0..threads - 1).map do |i| 
    Thread.new do 
    (i..messages * threads - 1).step(threads) do |n| 
     email = recipients_array[n].join(", ") 
     emails_handled[n] = 1 
     # ... do stuff with your email 
    end 
    end 
end 

• Каждый поток имеет один и тот же шаг и ту же конечную точку, но другую начальную точку, так что они не конфликтуют. Это не оптимально, но я довольно новичок в потоках.
• Если вы хотите получить recipient_count, вы можете позвонить по телефону emails_handled.compact.reduce(:+). Я не был уверен, нужен ли вам recipient_count для чего-либо кроме поиска recipients_array[] - если нет, вы можете полностью сбросить emails_handled.