Многопоточное программированиеВ этой главе
 Скачать 0.74 Mb.
|
|
Перенос приложения в версию Python 3 По аналогии с mtsleepF.py, candy.py представляет собой еще один пример того, что достаточно применить инструмент 2to3, чтобы сформировать работоспособную версию для Python 3 (для которой должно использоваться имя candy3.py). Оставля- ем проверку этого утверждения в качестве упражнения для читателя. Резюме В настоящей главе было продемонстрировано использование только некоторых примитивов синхронизации, которые входят в состав модуля threading. Поэто- му при желании читатель может найти для себя широкие возможности по иссле- дованию этой тематики. Однако следует учитывать, что объекты, представленные в 06_ch04.indd 224 22.01.2015 22:00:49 225 4.8. Проблема “производитель–потребитель” и модуль Queue/queue указанном модуле, полностью соответствуют своему определению, т.е. являются при- митивами. Это означает, что разработчик вполне может заняться созданием на их основе собственных классов и структур данных, обеспечив, в частности, их потоковую безопасность. Для этого в стандартной библиотеке Python предусмотрен еще один объект, Queue. 4.8. Проблема “производитель– потребитель” и модуль Queue/queue В заключительном примере иллюстрируется принцип работы “производитель– потребитель”, согласно которому производитель товаров или поставщик услуг про- изводит товары или подготавливает услуги и размещает их в структуре данных напо- добие очереди. Интервалы между отдельными событиями передачи произведенных товаров в очередь не детерминированы, как и интервалы потребления товаров. Модуль Queue (который носит это имя в версии Python 2.x, но переименован в queue в версии 3.x) предоставляет механизм связи между потоками, с по- мощью которого отдельные потоки могут совместно использовать данные. В данном случае создается очередь, в которую производитель (один поток) помещает новые товары, а потребитель (другой поток) их расходует. В табл. 4.5 перечислены различные атрибуты, которые представлены в указанном модуле. Таблица 4.5. Общие атрибуты модуля Queue/queue Атрибут Описание Классы модуля Queue/queue Queue(maxsize=0) Создает очередь с последовательной организацией, имеющую указанный размер maxsize, которая не позволяет вставлять новые блоки после достижения этого размера. Если размер не указан, то длина очереди становится неограниченной LifoQueue(maxsize=0) Создает стек, имеющий указанный размер maxsize, который не позволяет вставлять новые блоки после достижения этого размера. Если размер не указан, то длина стека становится нео- граниченной PriorityQueue(maxsize=0) Создает очередь по приоритету, имеющую указанный размер maxsize, которая не позволяет вставлять новые блоки после достижения этого размера. Если размер не указан, то длина очереди становится неограниченной Исключения модуля Queue/queue Empty Активизируется при вызове метода get*() применительно к пустой очереди Full Активизируется при вызове метода put*() применительно к заполненной очереди Методы объекта Queue/queue qsize() Возвращает размер очереди (это — приблизительное значение, поскольку при выполнении этого метода может происходить обновление очереди другими потоками) 06_ch04.indd 225 22.01.2015 22:00:49 Глава 4 Многопоточное программирование 226 Атрибут Описание empty() Возвращает True, если очередь пуста; в противном случае воз- вращает False full() Возвращает True, если очередь заполнена; в противном случае возвращает False put(item, block=True, timeout=None ) Помещает элемент item в очередь; если значение block равно True (по умолчанию) и значение timeout равно None, уста- навливает блокировку до тех пор, пока в очереди не появится свободное место. Если значение timeout является положи- тельным, блокирует очередь самое больше на timeout секунд, а если значение block равно False, активизирует исключение Empty put_nowait(item) То же, что и put(item, False) get(block=True, timeout=None) Получает элемент из очереди, если задано значение block (от- личное от 0); устанавливает блокировку до того времени, пока элемент не станет доступным get_nowait() То же, что и get(False) task_done() Используется для указания на то, что работа по постановке элемента в очередь завершена, в сочетании с описанным ниже методом join() join() Устанавливает блокировку до того времени, пока не будут обра- ботаны все элементы в очереди; сигнал об этом вырабатывается путем вызова описанного выше метода task_done() Для демонстрации того, как реализуется принцип “производитель–потребитель” с помощью модуля Queue/queue, воспользуемся примером 4.12 (prodcons.py). Ниже приведен вывод, полученный при одном запуске на выполнение этого сценария. $ prodcons.py starting writer at: Sun Jun 18 20:27:07 2006 producing object for Q... size now 1 starting reader at: Sun Jun 18 20:27:07 2006 consumed object from Q... size now 0 producing object for Q... size now 1 consumed object from Q... size now 0 producing object for Q... size now 1 producing object for Q... size now 2 producing object for Q... size now 3 consumed object from Q... size now 2 consumed object from Q... size now 1 writer finished at: Sun Jun 18 20:27:17 2006 consumed object from Q... size now 0 reader finished at: Sun Jun 18 20:27:25 2006 all DONE Пример 4.12. Пример реализации принципа “производитель–потребитель” ( prodcons.py) В этой реализации принципа “производитель–потребитель” используются объек- ты Queue и вырабатывается случайным образом количество произведенных (и потре- бленных) товаров. Производитель и потребитель моделируются с помощью потоков, действующих отдельно и одновременно. Окончание табл. 4.5 06_ch04.indd 226 22.01.2015 22:00:50 227 4.8. Проблема “производитель–потребитель” и модуль Queue/queue 1 #!/usr/bin/env python 2 3 from random import randint 4 from time import sleep 5 from Queue import Queue 6 from myThread import MyThread 7 8 def writeQ(queue): 9 print 'producing object for Q...', 10 queue.put('xxx', 1) 11 print "size now", queue.qsize() 12 13 def readQ(queue): 14 val = queue.get(1) 15 print 'consumed object from Q... size now', \ 16 queue.qsize() 17 18 def writer(queue, loops): 19 for i in range(loops): 20 writeQ(queue) 21 sleep(randint(1, 3)) 22 23 def reader(queue, loops): 24 for i in range(loops): 25 readQ(queue) 26 sleep(randint(2, 5)) 27 28 funcs = [writer, reader] 29 nfuncs = range(len(funcs)) 30 31 def main(): 32 nloops = randint(2, 5) 33 q = Queue(32) 34 35 threads = [] 36 for i in nfuncs: 37 t = MyThread(funcs[i], (q, nloops), 38 funcs[i].__name__) 39 threads.append(t) 40 41 for i in nfuncs: 42 threads[i].start() 43 44 for i in nfuncs: 45 threads[i].join() 46 47 print 'all DONE' 48 49 if __name__ == '__main__': 50 main() Вполне очевидно, что операции, выполняемые производителем и потребителем, не всегда чередуются, поскольку образуют две независимые последовательности. (Весьма удачно то, что в нашем распоряжении имеется готовый механизм выработки случайных чисел!) Если же говорить серьезно, то события, происходящие в действи- тельности, как правило, подчиняются законам случайности и недетерминированы. 06_ch04.indd 227 22.01.2015 22:00:50 Глава 4 Многопоточное программирование 228 Построчное объяснение Строки 1–6 В этом модуле используется объект Queue.Queue, а также потоки, сформирован- ные с помощью класса myThread.MyThread, как было описано ранее. Метод random. randint() применяется для внесения элемента случайности в операции производ- ства и потребления. (Заслуживает внимания то, что метод random.randint() дей- ствует точно так же, как и метод random.randrange(), но предусматривает включе- ние в интервал вырабатываемых случайных чисел начального и конечного значений.) Строки 8–16 Функции writeQ() и readQ() выполняют следующие операции: первая из них помещает объект в очередь (в качестве объекта может использоваться, например, строка 'xxx'), а вторая извлекает объект из очереди. Следует учитывать, что опера- ции постановки в очередь и изъятия из очереди осуществляются одновременно по отношению только к одному объекту. Строки 18–26 Метод writer() выполняется как отдельный поток, единственным назначением которого является выработка одного элемента для постановки в очередь, переход на время в состояние ожидания, а затем повтор этого цикла указанное количество раз, причем количество повторов устанавливается при выполнении сценария случайным образом. Метод reader() действует аналогично, если не считать того, что он не ста- вит, а извлекает элементы из очереди. Необходимо отметить, что устанавливаемая случайным образом продолжитель- ность приостановки метода-производителя в секундах, как правило, меньше по срав- нению с той продолжительностью, на которую приостанавливается метод-потреби- тель. Это сделано для того, чтобы метод-потребитель не мог предпринять попытки извлечения элементов из пустой очереди. Сокращение продолжительности прио- становки метода-производителя способствует повышению вероятности того, что в распоряжении метода-потребителя всегда будет пригодный для извлечения элемент, когда настанет время очередного выполнения этой операции. Строки 28-29 Это всего лишь подготовительные строки, с помощью которых задается общее ко- личество потоков, подлежащих порождению и запуску. Строки 31–47 Наконец, предусмотрена функция main(), которая должна выглядеть весьма по- добной функциям main() из всех прочих сценариев, приведенных в этой главе. Соз- даются необходимые потоки и осуществляется их запуск, а окончание работы насту- пает после того, как оба потока завершают свое выполнение. На основании этого примера можно сделать вывод, что программа, предназна- ченная для выполнения нескольких задач, может быть организована так, чтобы для реализации каждой из задач применялись отдельные потоки. Результатом может стать получение гораздо более наглядного проекта программы по сравнению с одно- поточной программой, в которой предпринимается попытка обеспечить выполнение всех задач. 06_ch04.indd 228 22.01.2015 22:00:50 229 4.9. Дополнительные сведения об использовании потоков В настоящей главе было показано, что применение однопоточного процесса мо- жет стать препятствием к повышению производительности приложения. Особенно значительно может быть повышена производительность программ, основанных на последовательном выполнении независимых, недетерминированных и не имеющих причинных зависимостей задач, в результате их разбиения на отдельные задачи, выполняемые отдельными потоками. Существенный выигрыш от перехода к мно- гопоточной обработке может быть достигнут не во всех приложениях. Причинами этого могут стать дополнительные издержки, а также тот факт, что сам интерпрета- тор Python представляет собой однопоточное приложение. Тем не менее овладение функциональными возможностями многопоточной организации Python позволяет взять этот инструмент на вооружение, когда это оправдано. 4.9. Дополнительные сведения об использовании потоков Прежде чем приступать к повсеместному применению средств поддержки мно- гопоточности, следует провести краткий обзор особенностей такой организации программирования. Вообще говоря, применение нескольких потоков в программе может способствовать ее улучшению. Однако в интерпретаторе Python применяется глобальная блокировка, которая накладывает свои ограничения, поэтому многопо- точная организация является более подходящей для приложений, ограничиваемых пропускной способностью ввода-вывода (при вводе-выводе происходит освобожде- ние глобальной блокировки интерпретатора, что способствует повышению степени распараллеливания), а не приложений, ограничиваемых пропускной способностью процессора. В последнем случае для достижения более высокой степени распаралле- ливания необходимо иметь возможность параллельного выполнения процессов не- сколькими ядрами или процессорами. Не вдаваясь в дополнительные подробности (поскольку некоторые из соответ- ствующих тем уже рассматривались в главе “Среда выполнения” книги “Core Python Programming” или “Core Python Language Fundamentals”), перечислим основные альтернативы модулю threading, касающиеся поддержки нескольких потоков или процессов. 4.9.1. Модуль subprocess В первую очередь вместо модуля threading можно применить модуль subprocess, когда возникает необходимость запуска новых процессов, либо для выполнения кода, либо для обеспечения обмена данными с другими процессами через стандартные файлы ввода-вывода (stdin, stdout, stderr). Этот модуль был введен в версии Python 2.4. 4.9.2. Модуль multiprocessing Этот модуль, впервые введенный в Python 2.6, позволяет запускать процессы для нескольких ядер или процессоров, но с интерфейсом, весьма напоми- нающим интерфейс модуля threading. Он также поддерживает различные механизмы передачи данных между процессами, применяемыми для вы- полнения совместной работы. 06_ch04.indd 229 22.01.2015 22:00:51 Глава 4 Многопоточное программирование 230 4.9.3. Модуль concurrent.futures Это новая высокоуровневая библиотека, которая работает только на уровне заданий. Это означает, что при использовании модуля concurrent.futures исключается необходимость заботиться о синхронизации либо управлять потоками или процессами. Достаточно лишь указать поток или пул процес- са с определенным количеством рабочих потоков, передать задания на вы- полнение и собрать полученные результаты. Этот модуль впервые появился в версии Python 3.2, но перенесен также в версию Python 2.6 и последующие версии. Модуль можно получить по адресу http://code.google.com/p/ pythonfutures. Рассмотрим вариант сценария bookrank3.py с указанными изменениями. При условии, что все прочее остается таким, как прежде, рассмотрим новые операторы импорта и изменившуюся часть сценария _main(): from concurrent.futures import ThreadPoolExecutor def _main(): print('At', ctime(), 'on Amazon...') with ThreadPoolExecutor(3) as executor: for isbn in ISBNs: executor.submit(_showRanking, isbn) print('all DONE at:', ctime()) Методу concurrent.futures.ThreadPoolExecutor передается параметр, пред- ставляющий собой размер пула потоков, а приложение применяется для определе- ния рангов трех книг. Безусловно, это — приложение, ограничиваемое пропускной способностью ввода-вывода, для которого применение потоков оказывает наиболь- шую пользу. Что касается приложений, ограничиваемых пропускной способно- стью процессора, то вместо указанного метода целесообразно было бы использовать concurrent.futures.ProcessPoolExecutor. После создания управляющего объекта (действие которого распространяется на потоки или процессы), отвечающего за планирование заданий и сбор результатов, можно вызвать его метод submit() для выполнения намеченной ранее задачи по- рождения потока. После полного переноса в версию Python 3 путем замены оператора формати- рования строки методом str.format(), повсеместного введения инструкции with и использования метода map() управляющего объекта появляется возможность полно- стью удалить метод _showRanking() и передать его функции в программу _main(). Заключительная версия сценария bookrank3CF.py приведена в примере 4.13. Пример 4.13. Применение средств управления заданиями высокого уровня ( bookrank3CF.py) В этом участке кода, как и в предыдущих примерах, осуществляется сбор с экрана данных о рангах книг, но на этот раз с помощью модуля concurrent.futures. 1 #!/usr/bin/env python 2 3 from concurrent.futures import ThreadPoolExecutor 4 from re import compile 06_ch04.indd 230 22.01.2015 22:00:51 231 4.9. Дополнительные сведения об использовании потоков 5 from time import ctime 6 from urllib.request import urlopen as uopen 7 8 REGEX = compile(b'#([\d,]+) in Books ') 9 AMZN = 'http://amazon.com/dp/' 10 ISBNs = { 11 '0132269937': 'Core Python Programming', 12 '0132356139': 'Python Web Development with Django', 13 '0137143419': 'Python Fundamentals', 14 } 15 16 def getRanking(isbn): 17 with uopen('{0}{1}'.format(AMZN, isbn)) as page: 18 return str(REGEX.findall(page.read())[0], 'utf-8') 19: 20: def _main(): 21: print('At', ctime(), 'on Amazon...') 22: with ThreadPoolExecutor(3) as executor: 23: for isbn, ranking in zip( 24: ISBNs, executor.map(getRanking, ISBNs)): 25: print('- %r ranked %s' % (ISBNs[isbn], ranking) 26: print('all DONE at:', ctime()) 27: 28: if __name__ == '__main__': 29: main() Построчное объяснение Строки 1–14 Если не считать новой инструкции import, то вся первая половина этого сцена- рия полностью идентична той, что приведена в файле bookrank3.py, который рас- сматривался выше в главе. Строки 16–18 В новой функции getRanking() используются инструкция with и функция str. format(). Аналогичные изменения можно внести в сценарий bookrank.py, посколь- ку оба указанных средства доступны также в версии 2.6 и последующих (а не пред- усмотрены исключительно в версиях 3.x). Строки 20–26 В предыдущем примере кода использовался метод executor.submit() для фор- мирования заданий. В данном примере предусмотрены некоторые изменения в свя- зи с использованием метода executor.map(), поскольку он позволяет реализовать функции из _showRanking() и полностью исключать их поддержку из нашего кода. Полученный вывод почти аналогичен тому, который рассматривался ранее: $ python3 bookrank3CF.py At Wed Apr 6 00:21:50 2011 on Amazon... - 'Core Python Programming' ranked 43,992 - 'Python Fundamentals' ranked 1,018,454 - 'Python Web Development with Django' ranked 502,566 all DONE at: Wed Apr 6 00:21:55 2011 06_ch04.indd 231 22.01.2015 22:00:51 Глава 4 Многопоточное программирование 232 Дополнительные сведения об истории создания модуля concurrent.futures можно найти с помощью приведенных ниже ссылок. http://docs.python.org/dev/py3k/library/concurrent.futures.html http://code.google.com/p/pythonfutures/ http://www.python.org/dev/peps/pep-3148/ Краткое описание этих параметров, а также другая информация, касающаяся мо- дулей и пакетов для многопоточной организации программы, приведена в следую- щем разделе. |