ThreadPoolExecutor Python – 03

Resumo : neste tutorial, você aprenderá como usar Python ThreadPoolExecutorpara desenvolver programas multithread.

Introdução à classe Python ThreadPoolExecutor

No tutorial multithreading , você aprendeu como gerenciar vários threads em um programa usando a Threadclasse do threadingmódulo. A Threadclasse é útil quando você deseja criar threads manualmente.

No entanto, gerenciar threads manualmente não é eficiente porque criar e destruir muitos threads frequentemente é muito caro em termos de custos computacionais.

Em vez de fazer isso, você pode reutilizar os threads se espera executar muitas tarefas ad-hoc no programa. Um pool de threads permite que você consiga isso.

Grupo de discussão

Um pool de threads é um padrão para obter simultaneidade de execução em um programa. Um pool de threads permite gerenciar automaticamente um pool de threads de forma eficiente:

Cada thread no pool é chamado de thread de trabalho ou trabalhador. Um pool de threads permite reutilizar os threads de trabalho assim que as tarefas forem concluídas. Também protege contra falhas inesperadas, como exceções .

Normalmente, um conjunto de encadeamentos permite configurar o número de encadeamentos de trabalho e fornece uma convenção de nomenclatura específica para cada encadeamento de trabalho.

Para criar um pool de threads, você usa a ThreadPoolExecutorclasse do concurrent.futuresmódulo.

ThreadPoolExecutor

A ThreadPoolExecutorclasse estende a Executorclasse e retorna um Futureobjeto.

Executor

A Executorclasse possui três métodos para controlar o pool de threads:

• submit()– despacha uma função para ser executada e retorna um Futureobjeto. O submit()método pega uma função e a executa de forma assíncrona.
• map()– executa uma função de forma assíncrona para cada elemento em um iterável.
• shutdown()– desligue o executor.

Quando você cria uma nova instância da ThreadPoolExecutorclasse, o Python inicia o arquivo Executor.

Depois de concluir o trabalho com o executor, você deve chamar explicitamente o shutdown()método para liberar o recurso mantido pelo executor. Para evitar chamar o shutdown()método explicitamente, você pode usar o gerenciador de contexto .

Objeto futuro

A Futureé um objeto que representa o resultado eventual de uma operação assíncrona. A classe Future possui dois métodos úteis:

• result()– retornar o resultado de uma operação assíncrona.
• exception()– retornar a exceção de uma operação assíncrona caso ocorra uma exceção.

Exemplos de ThreadPoolExecutor em Python

O programa a seguir usa um único thread:

from time import sleep, perf_counter

def task(id):
    print(f'Starting the task {id}...')
    sleep(1)
    return f'Done with task {id}'

start = perf_counter()

print(task(1))
print(task(2))

finish = perf_counter()

print(f"It took {finish-start} second(s) to finish.")Linguagem de código:  Python  ( python )

Saída:

Starting the task 1...
Done with task 1
Starting the task 2...
Done with task 2
It took 2.0144479 second(s) to finish.Linguagem de código:  Python  ( python )

Como funciona.

Primeiro, defina a task()função que leva cerca de um segundo para terminar. A task()função chama a sleep()função para simular um atraso:

def task(id):
    print(f'Starting the task {id}...')
    sleep(1)
    return f'Done with task {id}'Linguagem de código:  Python  ( python )

Segundo, chame a task()função duas vezes e imprima o resultado. Antes e depois de chamar a task()função, usamos o perf_counter()para medir o horário de início e término:

start = perf_counter()

print(task(1))
print(task(2))

finish = perf_counter()Linguagem de código:  Python  ( python )

Terceiro, imprima o tempo que o programa levou para ser executado:

print(f"It took {finish-start} second(s) to finish.")Linguagem de código:  Python  ( python )

Como a task()função leva um segundo, chamá-la duas vezes levará cerca de 2 segundos.

Usando o exemplo do método submit()

Para executar a task()função simultaneamente, você pode usar a ThreadPoolExecutorclasse:

from time import sleep, perf_counter
from concurrent.futures import ThreadPoolExecutor

def task(id):
    print(f'Starting the task {id}...')
    sleep(1)
    return f'Done with task {id}'

start = perf_counter()

with ThreadPoolExecutor() as executor:
    f1 = executor.submit(task, 1)
    f2 = executor.submit(task, 2)

    print(f1.result())
    print(f2.result())    

finish = perf_counter()

print(f"It took {finish-start} second(s) to finish.")Linguagem de código:  Python  ( python )

Saída:

Starting the task 1...
Starting the task 2...
Done with task 1
Done with task 2
It took 1.0177214 second(s) to finish.Linguagem de código:  Python  ( python )

A saída mostra que o programa demorou cerca de 1 segundo para terminar.

Como funciona (vamos nos concentrar na parte do pool de threads):

Primeiro, importe a ThreadPoolExecutorclasse do concurrent.futuresmódulo:

from concurrent.futures import ThreadPoolExecutorLinguagem de código:  Python  ( python )

Segundo, crie um pool de threads usando ThreadPoolExecutorum gerenciador de contexto:

with ThreadPoolExecutor() as executor:Linguagem de código:  Python  ( python )

Terceiro, chamar a task()função duas vezes, passando-a para o submit()método do executor:

with ThreadPoolExecutor() as executor:
    f1 = executor.submit(task, 1)
    f2 = executor.submit(task, 2)

    print(f1.result())
    print(f2.result()) Linguagem de código:  Python  ( python )

O submit()método retorna um objeto Future. Neste exemplo, temos dois objetos Future f1e f2. Para obter o resultado do objeto Future, chamamos seu result()método.

Usando o exemplo do método map()

O programa a seguir usa uma ThreadPoolExecutorclasse. Porém, em vez de usar o submit()método, ele usa o map()método para executar uma função:

from time import sleep, perf_counter
from concurrent.futures import ThreadPoolExecutor


def task(id):
    print(f'Starting the task {id}...')
    sleep(1)
    return f'Done with task {id}'

start = perf_counter()

with ThreadPoolExecutor() as executor:
    results = executor.map(task, [1,2])
    for result in results:
        print(result)

finish = perf_counter()

print(f"It took {finish-start} second(s) to finish.")Linguagem de código:  Python  ( python )

Como funciona.

Primeiro, chame o map()método do objeto executor para executar a função de tarefa para cada id na lista [1,2]. O map()método retorna um iterador que contém o resultado das chamadas de função.

results = executor.map(task, [1,2])Linguagem de código:  Python  ( python )

Segundo, repita os resultados e imprima-os:

for result in results:
    print(result)Linguagem de código:  Python  ( python )

Exemplo prático de Python ThreadPoolExecutor

O programa a seguir baixa várias imagens da Wikipedia usando um pool de threads:

from concurrent.futures import ThreadPoolExecutor
from urllib.request import urlopen
import time
import os

def download_image(url):
    image_data = None
    with urlopen(url) as f:
        image_data = f.read()

    if not image_data:
        raise Exception(f"Error: could not download the image from {url}")

    filename = os.path.basename(url)
    with open(filename, 'wb') as image_file:
        image_file.write(image_data)
        print(f'{filename} was downloaded...')

start = time.perf_counter()

urls = ['https://upload.wikimedia.org/wikipedia/commons/9/9d/Python_bivittatus_1701.jpg',
        'https://upload.wikimedia.org/wikipedia/commons/4/48/Python_Regius.jpg',
        'https://upload.wikimedia.org/wikipedia/commons/d/d3/Baby_carpet_python_caudal_luring.jpg',
        'https://upload.wikimedia.org/wikipedia/commons/f/f0/Rock_python_pratik.JPG',
        'https://upload.wikimedia.org/wikipedia/commons/0/07/Dulip_Wilpattu_Python1.jpg']

with ThreadPoolExecutor() as executor:
      executor.map(download_image, urls)

finish = time.perf_counter()    

print(f'It took {finish-start} second(s) to finish.')Linguagem de código:  Python  ( python )

Como funciona.

Primeiro, defina uma função download_image()que baixe uma imagem de uma URL e a salve em um arquivo:

def download_image(url):
    image_data = None
    with urlopen(url) as f:
        image_data = f.read()

    if not image_data:
        raise Exception(f"Error: could not download the image from {url}")

    filename = os.path.basename(url)
    with open(filename, 'wb') as image_file:
        image_file.write(image_data)
        print(f'{filename} was downloaded...')Linguagem de código:  Python  ( python )

A download_image()função a urlopen()função do urllib.requestmódulo para baixar uma imagem de um URL.

Segundo, execute a download_image()função usando um pool de threads chamando o map()método do ThreadPoolExecutorobjeto:

with ThreadPoolExecutor() as executor:
      executor.map(download_image, urls)Linguagem de código:  Python  ( python )

Resumo

• Um pool de threads é um padrão para gerenciar vários threads com eficiência.
• Use ThreadPoolExecutorclasse para gerenciar um pool de threads em Python.
• Chame o submit()método de ThreadPoolExecutorpara enviar uma tarefa ao conjunto de threads para execução. O submit()método retorna um objeto Future.
• Chame o map()método da ThreadPoolExecutorclasse para executar uma função em um pool de threads com cada elemento de uma lista.