Python'da Eş Zamanlılık - İş Parçacığı Havuzu

Çok iş parçacıklı görevlerimiz için çok sayıda iş parçacığı oluşturmamız gerektiğini varsayalım. Çok fazla iş parçacığı nedeniyle birçok performans sorunu olabileceğinden, hesaplama açısından en pahalı olacaktır. Verimliliğin sınırlanması önemli bir sorun olabilir. Bu sorunu bir iş parçacığı havuzu oluşturarak çözebiliriz. Bir iş parçacığı havuzu, iş verilmeye hazır duran, önceden başlatılmış ve boşta iş parçacıklarının grubu olarak tanımlanabilir. İş parçacığı havuzu oluşturmak, çok sayıda görev yapmamız gerektiğinde her görev için yeni iş parçacığı başlatmaya tercih edilir. Bir iş parçacığı havuzu, çok sayıda iş parçacığının eşzamanlı yürütülmesini aşağıdaki gibi yönetebilir -

Bir evre havuzundaki bir evre yürütmeyi tamamlarsa, o evre yeniden kullanılabilir.
Bir evre sonlandırılırsa, o evreyi değiştirmek için başka bir evre yaratılacaktır.

Python Modülü - Concurrent.futures

Python standart kitaplığı şunları içerir: concurrent.futuresmodül. Bu modül, geliştiricilere eşzamansız görevleri başlatmak için yüksek düzeyde bir arayüz sağlamak için Python 3.2'ye eklenmiştir. İş parçacığı veya süreç havuzunu kullanarak görevleri yürütmek için arayüz sağlamak için Python'un iş parçacığı ve çoklu işlem modüllerinin üstündeki bir soyutlama katmanıdır.

Sonraki bölümlerimizde, concurrent.futures modülünün farklı sınıflarını öğreneceğiz.

Yürütme Sınıfı

Executorsoyut bir sınıftır concurrent.futuresPython modülü. Doğrudan kullanılamaz ve aşağıdaki beton alt sınıflardan birini kullanmamız gerekir -

ThreadPoolExecutor
ProcessPoolExecutor

ThreadPoolExecutor - Beton Alt Sınıfı

Executor sınıfının somut alt sınıflarından biridir. Alt sınıf, çoklu iş parçacığı kullanır ve görevleri göndermek için bir iş parçacığı havuzu elde ederiz. Bu havuz, görevleri mevcut iş parçacıklarına atar ve çalıştırmaları için planlar.

ThreadPoolExecutor nasıl oluşturulur?

Yardımıyla concurrent.futures modül ve beton alt sınıfı Executor, kolayca bir iş parçacığı havuzu oluşturabiliriz. Bunun için bir inşa etmemiz gerekiyorThreadPoolExecutorhavuzda istediğimiz konu sayısı ile. Varsayılan olarak sayı 5'tir. Ardından iş parçacığı havuzuna bir görev gönderebiliriz. Biz ne zamansubmit() bir görev, geri alıyoruz Future. Future nesnesinin adı verilen bir yöntemi vardırdone(), geleceğin çözülüp çözülmediğini söyler. Bununla, gelecekteki bu belirli nesne için bir değer belirlendi. Bir görev bittiğinde, iş parçacığı havuzu yürütücüsü değeri gelecekteki nesneye ayarlar.

Misal

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

Çıktı

False
True
Completed

Yukarıdaki örnekte, bir ThreadPoolExecutor5 diş ile inşa edilmiştir. Daha sonra mesaj vermeden önce 2 saniye bekleyecek bir görev, iş parçacığı havuzu yürütücüsüne gönderilir. Çıktıdan görüldüğü gibi, görev 2 saniyeye kadar tamamlanmaz, bu nedenle ilk çağrıdone()False döndürür. 2 saniye sonra görev tamamlanır ve geleceğin sonucunu arayarak elde ederiz.result() yöntem.

ThreadPoolExecutor - Bağlam Yöneticisi

Örneklemenin başka bir yolu ThreadPoolExecutorbağlam yöneticisinin yardımıyla. Yukarıdaki örnekte kullanılan yönteme benzer şekilde çalışır. Bağlam yöneticisini kullanmanın temel avantajı, sözdizimsel olarak iyi görünmesidir. Örnekleme, aşağıdaki kod yardımıyla yapılabilir -

with ThreadPoolExecutor(max_workers = 5) as executor

Misal

Aşağıdaki örnek Python belgelerinden ödünç alınmıştır. Bu örnekte, her şeyden önceconcurrent.futuresmodülün içe aktarılması gerekiyor. Ardından adlı bir işlevload_url()istenen url'yi yükleyecek şekilde oluşturulur. İşlev daha sonra oluştururThreadPoolExecutorhavuzdaki 5 konu ile. ThreadPoolExecutorbağlam yöneticisi olarak kullanılmıştır. Geleceğin sonucunu arayarak alabiliriz.result() yöntem.

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

Çıktı

Aşağıdaki Python betiğinin çıktısı olacaktır -

'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

Executor.map () işlevinin kullanımı

Python map()işlevi, bir dizi görevde yaygın olarak kullanılmaktadır. Böyle bir görev, yinelenebilir öğelerdeki her öğeye belirli bir işlevi uygulamaktır. Benzer şekilde, bir yineleyicinin tüm öğelerini bir işleve eşleyebilir ve bunları bağımsız işler olarak gönderebiliriz.ThreadPoolExecutor. İşlevin nasıl çalıştığını anlamak için aşağıdaki Python betiği örneğini düşünün.

Misal

Aşağıdaki bu örnekte, harita işlevi, square() değerler dizisindeki her değer için işlev.

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

Çıktı

Yukarıdaki Python betiği aşağıdaki çıktıyı üretir -