DjangoBB LoFi version

Полная версия: Реализация параллельной загрузки картинок в Python

Начало » Python для новичков » Реализация параллельной загрузки картинок в Python

Spectral

Ноя. 8, 2011 17:33:50

Добрый вечер!

Подскажите пожалуйста, через какой модуль лучше всего реализовывать такую задачку :
есть список URL картинок, которые можно скачать, например, с помощью urlretrieve,
хочется, чтобы скрипт умел загружать картинки параллельно, примерно по такой схеме -
новую параллельную загрузку не начинать, пока не закончили скачиваться, например, 10 уже имеющихся загрузок.
Оптимально ли будет просто написать цикл по циклу, в котором стартуются функции загрузки и применён метод join (или joinall)?

Так вот, через какой модуль лучше всего это делать - Threading или subprocess?
Может ли кто-нибудь подкинуть какой-нибудь готовый пример, где реализована параллельная загрузка?
Благодарю за внимание.

s0rg

Ноя. 8, 2011 17:35:46

http://docs.python.org/library/multiprocessing.html

Ноя. 8, 2011 20:54:43

Потоков достаточно. Примерный алгоритм элементарен:
1. Создаёшь очередь Queue.
2. Заполняешь очередь (опционально, зависит от алгоритма).
3. Запускаешь 10 потоков, все они имеют один алгоритм.
4. Рабочий поток берёт из очереди задачу (если нет ничего – ждёт или или завершается, или завершается по спец. маркеру полученному из очереди) и выполняет её.
5. Главный поток пополняет очередь заданиями или нихрена не делает, ожидает завершения рабочих.

..bw

Spectral

Ноя. 10, 2011 21:01:22

Снова добрый вечер!
Спасибо bw за наводку к подходу.
Остановился на таком подходе (код не совсем рабочий):

from Queue import Queue
from threading import Thread

source = [obj1, obj2, obj3]
num_worker_threads = ...

def do_work(item):
     ...
     return result
  
def worker(q):
    while True:
        item = q.get()
        do_work(item)
        q.task_done()

q = Queue()
for i in range(num_worker_threads):
     t = Thread(target=worker, args=(q,))
     t.daemon = True
     t.start()

for item in source:
    q.put(item)

q.join()

Теперь применительно к своей задаче я написал ещё и внешний цикл ко всему этому (естественно, определения функций из него вынес),
чтобы формировать очереди снова и снова. Получится вполне рабочий скрипт, НО:
в конце концов скрипт с ошибкой “thread.error: can't start new thread”, сообщение об ошибке от треда с номером то ли за 600.
Может ли кто-нибудь подсказать, в чём дело или как нужно подправить скрипт?
Я думал, что метод q.join ожидает окончания всех загруженных тредов и очищается сам, а вместо этого он только наполняется
новыми тредами, несмотря на то, что q пересоздается внутри цикла. Как быть?
UPD.:
Хм, наверно из-за while True: тред и живет вечно)

UPD № 2.:
Да, функцию worker переписать вот так:

def worker(q):
        item = q.get()
        do_work(item)
        q.task_done()

Вопрос снимается! :)

Ноя. 13, 2011 08:55:22

> Хм, наверно из-за while True: тред и живет вечно)
> функцию worker переписать вот так
Плохо переписал, он там нужен.

Как я говорил, есть тысяча и один способ завершения рабочего потока, зависит от деталей, например от того как очередь пополняется:

1. Очередь заполнена до начала работы и если она опустошается, то это всё.

q = Queue.Queue()
while True:
  try:
    task = q.get_nowait()
  except Queue.Empty:
    break
  do_somthing(task)

2. Очередь будет пополняться задачами во время работы (используем некий маркер для выхода, “пустую” задачу; он пихается в очередь раздатчиком задач до тех пора пока очередь пуста и пока есть работающие потоки).

_go_home_yankee = object()
q = Queue.Queue()
while True:
  task = q.get()
  if task is _go_home_yankee:
    break
  do_somthing(task)

В общем это всё совершенно элементарно. И это всё без join'а, не люблю я его, да как и потоки :-).
Что касается task_done + join + daemon + while, почему не сработало не знаю, task_done и join никогда не пользовался, но while там нужен обязательно.

..bw