Concurrence en Python - Pool de threads

Supposons que nous devions créer un grand nombre de threads pour nos tâches multithread. Ce serait le plus coûteux en calcul car il peut y avoir de nombreux problèmes de performances, en raison d'un trop grand nombre de threads. Un problème majeur pourrait être la limitation du débit. Nous pouvons résoudre ce problème en créant un pool de threads. Un pool de threads peut être défini comme le groupe de threads pré-instanciés et inactifs, qui sont prêts à travailler. La création d'un pool de threads est préférable à l'instanciation de nouveaux threads pour chaque tâche lorsque nous devons effectuer un grand nombre de tâches. Un pool de threads peut gérer l'exécution simultanée d'un grand nombre de threads comme suit -

Si un thread d'un pool de threads termine son exécution, ce thread peut être réutilisé.
Si un thread est terminé, un autre thread sera créé pour remplacer ce thread.

Module Python - Concurrent.futures

La bibliothèque standard Python comprend concurrent.futuresmodule. Ce module a été ajouté dans Python 3.2 pour fournir aux développeurs une interface de haut niveau pour le lancement de tâches asynchrones. Il s'agit d'une couche d'abstraction au-dessus des modules de thread et de multiprocessus de Python pour fournir l'interface pour exécuter les tâches à l'aide d'un pool de threads ou de processus.

Dans nos sections suivantes, nous découvrirons les différentes classes du module concurrent.futures.

Classe d'exécuteur

Executorest une classe abstraite du concurrent.futuresModule Python. Il ne peut pas être utilisé directement et nous devons utiliser l'une des sous-classes concrètes suivantes -

ThreadPoolExecutor
ProcessPoolExecutor

ThreadPoolExecutor - Une sous-classe concrète

C'est l'une des sous-classes concrètes de la classe Executor. La sous-classe utilise le multi-threading et nous obtenons un pool de threads pour soumettre les tâches. Ce pool affecte des tâches aux threads disponibles et planifie leur exécution.

Comment créer un ThreadPoolExecutor?

Avec l'aide de concurrent.futures module et sa sous-classe concrète Executor, nous pouvons facilement créer un pool de threads. Pour cela, nous devons construire unThreadPoolExecutoravec le nombre de threads que nous voulons dans le pool. Par défaut, le nombre est 5. Ensuite, nous pouvons soumettre une tâche au pool de threads. quand noussubmit() une tâche, nous récupérons un Future. L'objet Future a une méthode appeléedone(), qui indique si l'avenir est résolu. Avec cela, une valeur a été définie pour cet objet futur particulier. Lorsqu'une tâche se termine, l'exécuteur du pool de threads définit la valeur sur l'objet futur.

Exemple

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()

Production

False
True
Completed

Dans l'exemple ci-dessus, un ThreadPoolExecutora été construit avec 5 fils. Ensuite, une tâche, qui attendra 2 secondes avant de donner le message, est soumise à l'exécuteur du pool de threads. Comme le montre la sortie, la tâche ne se termine pas avant 2 secondes, donc le premier appel àdone()retournera False. Au bout de 2 secondes, la tâche est terminée et nous obtenons le résultat du futur en appelant leresult() méthode là-dessus.

Instanciation de ThreadPoolExecutor - Gestionnaire de contexte

Une autre façon d'instancier ThreadPoolExecutorest avec l'aide du gestionnaire de contexte. Cela fonctionne de manière similaire à la méthode utilisée dans l'exemple ci-dessus. Le principal avantage de l'utilisation du gestionnaire de contexte est qu'il a une bonne syntaxe. L'instanciation peut être effectuée à l'aide du code suivant -

with ThreadPoolExecutor(max_workers = 5) as executor

Exemple

L'exemple suivant est emprunté à la documentation Python. Dans cet exemple, tout d'abord leconcurrent.futuresle module doit être importé. Puis une fonction nomméeload_url()est créé qui chargera l'url demandée. La fonction crée alorsThreadPoolExecutoravec les 5 fils dans la piscine. leThreadPoolExecutora été utilisé comme gestionnaire de contexte. Nous pouvons obtenir le résultat du futur en appelant leresult() méthode là-dessus.

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)))

Production

Voici la sortie du script Python ci-dessus -

'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

Utilisation de la fonction Executor.map ()

Le Python map()La fonction est largement utilisée dans un certain nombre de tâches. L'une de ces tâches consiste à appliquer une certaine fonction à chaque élément des itérables. De même, nous pouvons mapper tous les éléments d'un itérateur à une fonction et les soumettre en tant que jobs indépendants à outThreadPoolExecutor. Prenons l'exemple suivant de script Python pour comprendre le fonctionnement de la fonction.

Exemple

Dans cet exemple ci-dessous, la fonction de carte est utilisée pour appliquer le square() fonction à chaque valeur du tableau de valeurs.

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()

Production

Le script Python ci-dessus génère la sortie suivante -