Simultaneidade em Python - Pool de Threads

Suponha que tenhamos que criar um grande número de threads para nossas tarefas multithread. Seria mais caro do ponto de vista computacional, pois pode haver muitos problemas de desempenho devido a muitos threads. Um grande problema pode ser a limitação do rendimento. Podemos resolver esse problema criando um pool de threads. Um pool de threads pode ser definido como o grupo de threads pré-instanciados e ociosos, que estão prontos para trabalhar. A criação de pool de threads é preferível a instanciar novos threads para cada tarefa quando precisamos realizar um grande número de tarefas. Um pool de threads pode gerenciar a execução simultânea de um grande número de threads da seguinte forma -

Se um encadeamento em um pool de encadeamentos conclui sua execução, esse encadeamento pode ser reutilizado.
Se um encadeamento for encerrado, outro encadeamento será criado para substituí-lo.

Módulo Python - Concurrent.futures

A biblioteca padrão do Python inclui o concurrent.futuresmódulo. Este módulo foi adicionado ao Python 3.2 para fornecer aos desenvolvedores uma interface de alto nível para iniciar tarefas assíncronas. É uma camada de abstração no topo dos módulos de threading e multiprocessamento do Python para fornecer a interface para executar as tarefas usando pool de thread ou processos.

Em nossas seções subsequentes, aprenderemos sobre as diferentes classes do módulo concurrent.futures.

Classe Executor

Executoré uma classe abstrata de concurrent.futuresMódulo Python. Não pode ser usado diretamente e precisamos usar uma das seguintes subclasses concretas -

ThreadPoolExecutor
ProcessPoolExecutor

ThreadPoolExecutor - uma subclasse concreta

É uma das subclasses concretas da classe Executor. A subclasse usa multi-threading e obtemos um pool de threads para enviar as tarefas. Este pool atribui tarefas aos threads disponíveis e os programa para execução.

Como criar um ThreadPoolExecutor?

Com a ajuda de concurrent.futures módulo e sua subclasse concreta Executor, podemos criar facilmente um pool de threads. Para isso, precisamos construir umThreadPoolExecutorcom o número de threads que queremos no pool. Por padrão, o número é 5. Em seguida, podemos enviar uma tarefa para o pool de threads. Quando nóssubmit() uma tarefa, voltamos a Future. O objeto Future tem um método chamadodone(), que informa se o futuro foi resolvido. Com isso, um valor foi definido para esse objeto futuro específico. Quando uma tarefa termina, o executor do pool de threads define o valor para o objeto futuro.

Exemplo

from concurrent.futures import ThreadPoolExecutor
from time import sleep
def task(message):
   sleep(2)
   return message

def main():
   executor = ThreadPoolExecutor(5)
   future = executor.submit(task, ("Completed"))
   print(future.done())
   sleep(2)
   print(future.done())
   print(future.result())
if __name__ == '__main__':
main()

Resultado

False
True
Completed

No exemplo acima, um ThreadPoolExecutorfoi construído com 5 fios. Em seguida, uma tarefa, que aguardará 2 segundos antes de dar a mensagem, é enviada ao executor do pool de threads. Como pode ser visto na saída, a tarefa não é concluída até 2 segundos, portanto, a primeira chamada paradone()retornará False. Após 2 segundos, a tarefa está concluída e obtemos o resultado do futuro chamando oresult() método sobre ele.

Instanciando ThreadPoolExecutor - Gerenciador de Contexto

Outra forma de instanciar ThreadPoolExecutoré com a ajuda do gerenciador de contexto. Funciona de forma semelhante ao método usado no exemplo acima. A principal vantagem de usar o gerenciador de contexto é que ele parece sintaticamente bom. A instanciação pode ser feita com a ajuda do seguinte código -

with ThreadPoolExecutor(max_workers = 5) as executor

Exemplo

O exemplo a seguir foi emprestado dos documentos do Python. Neste exemplo, em primeiro lugar, oconcurrent.futuresmódulo tem que ser importado. Então, uma função chamadaload_url()é criado, o que carregará o url solicitado. A função então criaThreadPoolExecutor com os 5 threads no pool. oThreadPoolExecutorfoi utilizado como gerenciador de contexto. Podemos obter o resultado do futuro chamando oresult() método sobre ele.

import concurrent.futures
import urllib.request

URLS = ['http://www.foxnews.com/',
   'http://www.cnn.com/',
   'http://europe.wsj.com/',
   'http://www.bbc.co.uk/',
   'http://some-made-up-domain.com/']

def load_url(url, timeout):
   with urllib.request.urlopen(url, timeout = timeout) as conn:
   return conn.read()

with concurrent.futures.ThreadPoolExecutor(max_workers = 5) as executor:

   future_to_url = {executor.submit(load_url, url, 60): url for url in URLS}
   for future in concurrent.futures.as_completed(future_to_url):
   url = future_to_url[future]
   try:
      data = future.result()
   except Exception as exc:
      print('%r generated an exception: %s' % (url, exc))
   else:
      print('%r page is %d bytes' % (url, len(data)))

Resultado

A seguir está a saída do script Python acima -

'http://some-made-up-domain.com/' generated an exception: <urlopen error [Errno 11004] getaddrinfo failed>
'http://www.foxnews.com/' page is 229313 bytes
'http://www.cnn.com/' page is 168933 bytes
'http://www.bbc.co.uk/' page is 283893 bytes
'http://europe.wsj.com/' page is 938109 bytes

Uso da função Executor.map ()

O Python map()função é amplamente utilizada em várias tarefas. Uma dessas tarefas é aplicar uma determinada função a cada elemento dentro dos iteráveis. Da mesma forma, podemos mapear todos os elementos de um iterador para uma função e enviá-los como trabalhos independentes paraThreadPoolExecutor. Considere o seguinte exemplo de script Python para entender como a função funciona.

Exemplo

Neste exemplo abaixo, a função de mapa é usada para aplicar o square() função para cada valor na matriz de valores.

from concurrent.futures import ThreadPoolExecutor
from concurrent.futures import as_completed
values = [2,3,4,5]
def square(n):
   return n * n
def main():
   with ThreadPoolExecutor(max_workers = 3) as executor:
      results = executor.map(square, values)
for result in results:
      print(result)
if __name__ == '__main__':
   main()

Resultado

O script Python acima gera a seguinte saída -