Файл: Многопоточное программированиеВ этой главе.pdf

223 4.7. Практическое применение многопоточной обработки
Строки 12–21
Оператор, обслуживающий эти вымышленные торговые автоматы, время от вре- мени пополняет запасы товара. Для моделирования этого действия применяется функция refill(). Вся процедура представляет критический участок кода; именно поэтому захват блокировки становится обязательным условием, при котором могут быть беспрепятственно выполнены все строки от начала до конца. В коде предусмо- трен вывод журнала со сведениями о выполненных действиях, с которым может оз- накомиться пользователь. Кроме того, вырабатывается предупреждение при попытке превысить максимально допустимый объем запасов (строки 17-18).
Строки 23–30
Функция buy() противоположна по своему назначению функции refill(); она позволяет покупателю каждый раз получать по одной единице из хранимых запасов.
Для обнаружения того, не исчерпаны ли уже конечные ресурсы, применяется услов- ное выражение (строка 26). Значение счетчика ни в коем случае не может стать отри- цательным, поэтому после достижения нулевого значения количества запасов вызов функции buy() блокируется до того момента, когда значение счетчика будет снова увеличено. Если этой функции передается значение неблокирующего флага, False, то вместо блокирования функция возвращает False, указывая на то, что ресурсы ис- черпаны.
Строки 32–40
Функции producer() и consumer(), моделирующие пополнение и потребление запаса конфет, просто повторяются в цикле и обеспечивают выполнение соответству- ющих вызовов функций refill() и buy(), приостанавливаясь на короткое время между вызовами.
Строки 42–55
В последней части кода сценария содержится вызов функции _main(), выпол- няемый при запуске сценария из командной строки, предусмотрена регистрация функции выхода, а также приведено определение функции _main(), в которой пред- усмотрена инициализация вновь созданной пары потоков, которые моделируют по- полнение и потребление конфет.
В коде создания потока, который представляет покупателя (потребителя), допол- нительно предусмотрена возможность вводить установленное случайным образом положительное смещение, при котором покупатель фактически может попытаться получить больше шоколадных батончиков, чем заправлено в торговый автомат по- ставщиком/производителем (в противном случае в коде никогда не возникла бы си- туация, в которой покупателю разрешалось бы совершать попытку купить шоколад- ный батончик из пустого автомата).
Выполнение сценарием приводит к получению примерно такого вывода:
$ python candy.py starting at: Mon Apr 4 00:56:02 2011
THE CANDY MACHINE (full with 5 bars)!
Buying candy... OK
Refilling candy... OK
Refilling candy... full, skipping
Buying candy... OK
06_ch04.indd 223 22.01.2015 22:00:49

Глава 4

Многопоточное программирование
224
Buying candy... OK
Refilling candy... OK
Buying candy... OK
Buying candy... OK
Buying candy... OK all DONE at: Mon Apr 4 00:56:08 2011
В процессе отладки этого сценария может потребоваться вмешательство
вручную
Если в какой-то момент нужно будет приступить к отладке сценария, в котором ис- пользуются семафоры, прежде всего необходимо точно знать, какое значение находит- ся в счетчике семафора в любое заданное время. В одном из упражнений в конце дан- ной главы предлагается реализовать такое решение в сценарии candy.py, возможно, переименовав его в candydebug.py, и предусмотреть в нем возможность отображать значение счетчика. Для этого может потребоваться рассмотреть исходный код модуля threading.py (причем, вероятно, и в версии Python 2, и в версии Python 3).
При этом следует учитывать, что имена примитивов синхронизации модуля threading не являются именами классов, даже притом, что в них используется вы- деление прописными буквами, как в ВерблюжьемСтиле, что делает их похожими на классы. В действительности эти примитивы представляют собой всего лишь одностроч- ные функции, применяемые для порождения необходимых объектов. При работе с мо- дулем threading необходимо учитывать две сложности: во-первых, невозможно поро- ждать подклассы создаваемых объектов (поскольку они представляют собой функции), а во-вторых, при переходе от версий 2.x к версиям 3.x изменились имена переменных.
Обе эти сложности можно было бы легко преодолеть, если бы в обеих версиях объек- ты предоставляли свободный и единообразный доступ к счетчику, что в действитель- ности не обеспечивается. Непосредственный доступ к значению счетчика может быть получен, поскольку он представляет собой всего лишь атрибут класса, но, как уже было сказано, имя переменной self.__value изменилось. Это означает, что переменная self._Semaphore__value в Python 2 была переименована в self._value в Python 3.
Что касается разработчиков, то наиболее удобный вариант применения интерфейса прикладного программирования API (по крайней мере, по мнению автора) состоит в создании подкласса класса threading._BoundedSemaphore и реализации метода
__len__(). Но если планируется поддерживать сценарий и в версии 2.x, и в версии 3.x, то следует использовать правильное значение счетчика, как было только что описано.

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