Взаимосвязь потоков
В реальной жизни, если группа людей работает над общей задачей, между ними должна быть связь для правильного выполнения задачи. Та же аналогия применима и к потокам. В программировании, чтобы сократить идеальное время процессора, мы создаем несколько потоков и назначаем разные подзадачи каждому потоку. Следовательно, должно быть средство связи, и они должны взаимодействовать друг с другом, чтобы завершить работу синхронно.
Рассмотрим следующие важные моменты, связанные с взаимодействием потоков:
No performance gain - Если мы не можем обеспечить надлежащую связь между потоками и процессами, тогда выигрыш в производительности от параллелизма и параллелизма будет бесполезен.
Accomplish task properly - Без надлежащего механизма взаимодействия между потоками поставленная задача не может быть выполнена должным образом.
More efficient than inter-process communication - Межпоточная связь более эффективна и проста в использовании, чем межпроцессная связь, поскольку все потоки в рамках процесса используют одно и то же адресное пространство, и им не нужно использовать общую память.
Структуры данных Python для потоковой связи
Многопоточный код сталкивается с проблемой передачи информации из одного потока в другой. Стандартные примитивы связи не решают эту проблему. Следовательно, нам необходимо реализовать наш собственный составной объект, чтобы разделять объекты между потоками, чтобы сделать коммуникацию потокобезопасной. Ниже приведены несколько структур данных, которые обеспечивают потокобезопасную связь после внесения в них некоторых изменений.
Наборы
Для использования структуры данных набора в потокобезопасном режиме нам необходимо расширить класс набора для реализации нашего собственного механизма блокировки.
пример
Вот пример расширения класса Python -
class extend_class(set):
def __init__(self, *args, **kwargs):
self._lock = Lock()
super(extend_class, self).__init__(*args, **kwargs)
def add(self, elem):
self._lock.acquire()
try:
super(extend_class, self).add(elem)
finally:
self._lock.release()
def delete(self, elem):
self._lock.acquire()
try:
super(extend_class, self).delete(elem)
finally:
self._lock.release()
В приведенном выше примере объект класса с именем extend_class был определен, который в дальнейшем унаследован от Python set class. В конструкторе этого класса создается объект блокировки. Теперь есть две функции -add() а также delete(). Эти функции определены и ориентированы на многопоточность. Они оба полагаются наsuper функциональность класса с одним ключевым исключением.
Декоратор
Это еще один ключевой метод поточно-ориентированного взаимодействия - использование декораторов.
пример
Рассмотрим пример Python, который показывает, как использовать декораторы & mminus;
def lock_decorator(method):
def new_deco_method(self, *args, **kwargs):
with self._lock:
return method(self, *args, **kwargs)
return new_deco_method
class Decorator_class(set):
def __init__(self, *args, **kwargs):
self._lock = Lock()
super(Decorator_class, self).__init__(*args, **kwargs)
@lock_decorator
def add(self, *args, **kwargs):
return super(Decorator_class, self).add(elem)
@lock_decorator
def delete(self, *args, **kwargs):
return super(Decorator_class, self).delete(elem)
В приведенном выше примере был определен метод декоратора с именем lock_decorator, который в дальнейшем наследуется от класса методов Python. Затем в конструкторе этого класса создается объект блокировки. Теперь есть две функции - add () и delete (). Эти функции определены и ориентированы на многопоточность. Оба они полагаются на функциональность суперкласса, за одним ключевым исключением.
Списки
Структура данных списка является поточно-ориентированной, быстрой и простой структурой для временного хранения в памяти. В Cpython GIL защищает от одновременного доступа к ним. Как мы узнали, списки потокобезопасны, но как насчет данных, которые в них хранятся? Собственно, данные списка не защищены. Например,L.append(x)не гарантирует возврата ожидаемого результата, если другой поток пытается сделать то же самое. Это потому, что, хотяappend() является атомарной операцией и потокобезопасной, но другой поток пытается изменить данные списка одновременно, поэтому мы можем видеть побочные эффекты состояний гонки на выходе.
Чтобы решить эту проблему и безопасно изменить данные, мы должны реализовать надлежащий механизм блокировки, который дополнительно гарантирует, что несколько потоков не могут потенциально работать в условиях гонки. Чтобы реализовать правильный механизм блокировки, мы можем расширить класс, как мы это делали в предыдущих примерах.
Некоторые другие атомарные операции со списками следующие:
L.append(x)
L1.extend(L2)
x = L[i]
x = L.pop()
L1[i:j] = L2
L.sort()
x = y
x.field = y
D[x] = y
D1.update(D2)
D.keys()
Здесь -
- L, L1, L2 все списки
- D, D1, D2 диктуют
- x, y - объекты
- я, j целые
Очереди
Если данные списка не защищены, нам, возможно, придется столкнуться с последствиями. Мы можем получить или удалить неверный элемент данных условий гонки. Поэтому рекомендуется использовать структуру данных очереди. Реальным примером очереди может быть однополосная дорога с односторонним движением, когда автомобиль входит первым, а выезжает первым. Более реальные примеры можно увидеть в очередях у билетных касс и на автобусных остановках.
Очереди по умолчанию представляют собой поточно-ориентированную структуру данных, и нам не нужно беспокоиться о реализации сложного механизма блокировки. Python предоставляет нам
Типы очередей
В этом разделе мы узнаем о разных типах очередей. Python предоставляет три варианта использования очередей из<queue> модуль -
- Обычные очереди (FIFO, первым пришел - первым ушел)
- LIFO, последний пришел - первым ушел
- Priority
Мы узнаем о различных очередях в следующих разделах.
Обычные очереди (FIFO, первым пришел - первым ушел)
Это наиболее часто используемые реализации очереди, предлагаемые Python. В этом механизме очередей тот, кто придет первым, первым получит услугу. FIFO также называют обычными очередями. Очереди FIFO могут быть представлены следующим образом -
Реализация очереди FIFO в Python
В Python очередь FIFO может быть реализована как с одним потоком, так и с многопоточностью.
Очередь FIFO с одним потоком
Для реализации очереди FIFO с одним потоком Queueclass реализует базовый контейнер "первым пришел - первым ушел". Элементы будут добавлены к одному «концу» последовательности с помощьюput(), и удален с другого конца с помощью get().
пример
Ниже приведена программа Python для реализации очереди FIFO с одним потоком.
import queue
q = queue.Queue()
for i in range(8):
q.put("item-" + str(i))
while not q.empty():
print (q.get(), end = " ")
Выход
item-0 item-1 item-2 item-3 item-4 item-5 item-6 item-7
Выходные данные показывают, что вышеуказанная программа использует один поток, чтобы проиллюстрировать, что элементы удаляются из очереди в том же порядке, в котором они вставляются.
Очередь FIFO с несколькими потоками
Для реализации FIFO с несколькими потоками нам необходимо определить функцию myqueue (), которая расширяется из модуля очереди. Работа методов get () и put () такая же, как описано выше при реализации очереди FIFO с одним потоком. Затем, чтобы сделать его многопоточным, нам нужно объявить и создать экземпляры потоков. Эти потоки будут использовать очередь в режиме FIFO.
пример
Ниже приведена программа Python для реализации очереди FIFO с несколькими потоками.
import threading
import queue
import random
import time
def myqueue(queue):
while not queue.empty():
item = queue.get()
if item is None:
break
print("{} removed {} from the queue".format(threading.current_thread(), item))
queue.task_done()
time.sleep(2)
q = queue.Queue()
for i in range(5):
q.put(i)
threads = []
for i in range(4):
thread = threading.Thread(target=myqueue, args=(q,))
thread.start()
threads.append(thread)
for thread in threads:
thread.join()
Выход
<Thread(Thread-3654, started 5044)> removed 0 from the queue
<Thread(Thread-3655, started 3144)> removed 1 from the queue
<Thread(Thread-3656, started 6996)> removed 2 from the queue
<Thread(Thread-3657, started 2672)> removed 3 from the queue
<Thread(Thread-3654, started 5044)> removed 4 from the queue
LIFO, очередь "последний в первом ушел"
Эта очередь использует полностью противоположную аналогию, чем очереди FIFO (First in First Out). В этом механизме очередей тот, кто идет последним, получит обслуживание первым. Это похоже на реализацию структуры данных стека. Очереди LIFO полезны при реализации поиска в глубину, как алгоритмы искусственного интеллекта.
Реализация очереди LIFO в Python
В python очередь LIFO может быть реализована как с одним потоком, так и с многопоточностью.
Очередь LIFO с одним потоком
Для реализации очереди LIFO с одним потоком Queue class будет реализовывать базовый контейнер «последним пришел - первым ушел», используя структуру Queue.LifoQueue. Теперь по звонкуput(), элементы добавляются в головку контейнера и удаляются из головки также при использовании get().
пример
Ниже приведена программа Python для реализации очереди LIFO с одним потоком.
import queue
q = queue.LifoQueue()
for i in range(8):
q.put("item-" + str(i))
while not q.empty():
print (q.get(), end=" ")
Output:
item-7 item-6 item-5 item-4 item-3 item-2 item-1 item-0
Выходные данные показывают, что приведенная выше программа использует один поток, чтобы проиллюстрировать, что элементы удаляются из очереди в порядке, обратном их вставке.
Очередь LIFO с несколькими потоками
Реализация аналогична тому, как мы реализовали очереди FIFO с несколькими потоками. Единственная разница в том, что нам нужно использоватьQueue класс, который будет реализовывать базовый контейнер «последним пришел - первым ушел», используя структуру Queue.LifoQueue.
пример
Ниже приведена программа Python для реализации очереди LIFO с несколькими потоками.
import threading
import queue
import random
import time
def myqueue(queue):
while not queue.empty():
item = queue.get()
if item is None:
break
print("{} removed {} from the queue".format(threading.current_thread(), item))
queue.task_done()
time.sleep(2)
q = queue.LifoQueue()
for i in range(5):
q.put(i)
threads = []
for i in range(4):
thread = threading.Thread(target=myqueue, args=(q,))
thread.start()
threads.append(thread)
for thread in threads:
thread.join()
Выход
<Thread(Thread-3882, started 4928)> removed 4 from the queue
<Thread(Thread-3883, started 4364)> removed 3 from the queue
<Thread(Thread-3884, started 6908)> removed 2 from the queue
<Thread(Thread-3885, started 3584)> removed 1 from the queue
<Thread(Thread-3882, started 4928)> removed 0 from the queue
Приоритетная очередь
В очередях FIFO и LIFO порядок элементов связан с порядком вставки. Однако во многих случаях приоритет важнее порядка вставки. Давайте рассмотрим пример из реального мира. Допустим, охрана в аэропорту проверяет людей разных категорий. Люди из VVIP, сотрудники авиакомпании, таможенник, категории могут проверяться по приоритету, а не по прибытию, как это происходит с простыми людьми.
Еще один важный аспект, который необходимо учитывать при выборе очереди с приоритетами, - это разработка планировщика задач. Один из распространенных способов - обслуживать большую часть задач агента в очереди на основе приоритета. Эта структура данных может использоваться для выбора элементов из очереди на основе их значения приоритета.
Реализация очереди приоритетов в Python
В python приоритетная очередь может быть реализована как с однопоточным, так и с многопоточным режимом.
Очередь приоритета с одним потоком
Для реализации приоритетной очереди с одним потоком Queue класс будет реализовывать задачу в приоритетном контейнере, используя структуру Queue.PriorityQueue. Теперь по звонкуput(), элементы добавляются со значением, где наименьшее значение будет иметь наивысший приоритет и, следовательно, извлекаются первыми с помощью get().
пример
Рассмотрим следующую программу Python для реализации очереди Priority с одним потоком -
import queue as Q
p_queue = Q.PriorityQueue()
p_queue.put((2, 'Urgent'))
p_queue.put((1, 'Most Urgent'))
p_queue.put((10, 'Nothing important'))
prio_queue.put((5, 'Important'))
while not p_queue.empty():
item = p_queue.get()
print('%s - %s' % item)
Выход
1 – Most Urgent
2 - Urgent
5 - Important
10 – Nothing important
В приведенном выше выводе мы видим, что в очереди хранятся элементы на основе приоритета - меньшее значение имеет высокий приоритет.
Приоритетная очередь с несколькими потоками
Реализация аналогична реализации очередей FIFO и LIFO с несколькими потоками. Единственная разница в том, что нам нужно использоватьQueue класс для инициализации приоритета с помощью структуры Queue.PriorityQueue. Другое отличие заключается в способе создания очереди. В приведенном ниже примере он будет создан с двумя идентичными наборами данных.
пример
Следующая программа Python помогает в реализации очереди приоритетов с несколькими потоками:
import threading
import queue
import random
import time
def myqueue(queue):
while not queue.empty():
item = queue.get()
if item is None:
break
print("{} removed {} from the queue".format(threading.current_thread(), item))
queue.task_done()
time.sleep(1)
q = queue.PriorityQueue()
for i in range(5):
q.put(i,1)
for i in range(5):
q.put(i,1)
threads = []
for i in range(2):
thread = threading.Thread(target=myqueue, args=(q,))
thread.start()
threads.append(thread)
for thread in threads:
thread.join()
Выход
<Thread(Thread-4939, started 2420)> removed 0 from the queue
<Thread(Thread-4940, started 3284)> removed 0 from the queue
<Thread(Thread-4939, started 2420)> removed 1 from the queue
<Thread(Thread-4940, started 3284)> removed 1 from the queue
<Thread(Thread-4939, started 2420)> removed 2 from the queue
<Thread(Thread-4940, started 3284)> removed 2 from the queue
<Thread(Thread-4939, started 2420)> removed 3 from the queue
<Thread(Thread-4940, started 3284)> removed 3 from the queue
<Thread(Thread-4939, started 2420)> removed 4 from the queue
<Thread(Thread-4940, started 3284)> removed 4 from the queue