Jak korzystać z wątków w Pythonie?

1278

Próbuję zrozumieć wątki w Pythonie. Przejrzałem dokumentację i przykłady, ale szczerze mówiąc, wiele przykładów jest zbyt wyrafinowanych i mam problem z ich zrozumieniem.

Jak wyraźnie pokazać zadania dzielone na wiele wątków?

albruno
źródło
31
Dobrą ogólną dyskusję na ten temat można znaleźć w najtrudniejszym problemie Pythona autorstwa Jeffa Knuppa. Podsumowując, wydaje się, że wątki nie są dla początkujących.
Matthew Walker,
111
haha, myślę, że wątki są dla wszystkich, ale początkujący nie są dla wątków :)))))
Bohdan
42
Wystarczy zaznaczyć, że ludzie powinni przeczytać wszystkie odpowiedzi, ponieważ późniejsze są prawdopodobnie lepsze, ponieważ nowe funkcje językowe są wykorzystywane ...
Gwyn Evans
5
Pamiętaj, aby napisać swoją podstawową logikę w C i wywołać ją za pomocą ctypów, aby naprawdę skorzystać z wątków Pythona.
aaa90210
4
Chciałem tylko dodać, że PyPubSub to świetny sposób na wysyłanie i odbieranie wiadomości kontrolujących przepływ wątków
ytpillai

Odpowiedzi:

1416

Ponieważ pytanie to zostało zadane w 2010 r., Wprowadzono proste uproszczenie sposobu wykonywania wielowątkowości za pomocą Pythona z mapą i pulą .

Poniższy kod pochodzi z artykułu / postu na blogu, który zdecydowanie powinieneś sprawdzić (bez powiązania) - Równoległość w jednym wierszu: lepszy model codziennych zadań gwintowania . Podsumuję poniżej - ostatecznie jest to tylko kilka wierszy kodu:

from multiprocessing.dummy import Pool as ThreadPool
pool = ThreadPool(4)
results = pool.map(my_function, my_array)

Która jest wielowątkową wersją:

results = []
for item in my_array:
    results.append(my_function(item))

Opis

Mapa jest fajną małą funkcją i kluczem do łatwego wprowadzania równoległości do kodu Pythona. Dla nieznajomych mapa jest czymś wyniesionym z funkcjonalnych języków, takich jak Lisp. Jest to funkcja, która odwzorowuje inną funkcję w sekwencji.

Mapa obsługuje dla nas iterację w sekwencji, stosuje funkcję i przechowuje wszystkie wyniki w przydatnej liście na końcu.

Wpisz opis zdjęcia tutaj


Realizacja

Równoległe wersje funkcji mapy zapewniają dwie biblioteki: wieloprocesowe, a także jej mało znane, ale równie fantastyczne dziecko krokowe: multiprocessing.dummy.

multiprocessing.dummyjest dokładnie taki sam jak moduł wieloprocesowy, ale zamiast tego używa wątków ( ważne rozróżnienie - użyj wielu procesów do zadań wymagających dużej mocy procesora; wątki do (i podczas) operacji we / wy :

multiprocessing.dummy replikuje interfejs API wieloprocesowości, ale jest niczym więcej jak opakowaniem wokół modułu wątków.

import urllib2
from multiprocessing.dummy import Pool as ThreadPool

urls = [
  'http://www.python.org',
  'http://www.python.org/about/',
  'http://www.onlamp.com/pub/a/python/2003/04/17/metaclasses.html',
  'http://www.python.org/doc/',
  'http://www.python.org/download/',
  'http://www.python.org/getit/',
  'http://www.python.org/community/',
  'https://wiki.python.org/moin/',
]

# Make the Pool of workers
pool = ThreadPool(4)

# Open the URLs in their own threads
# and return the results
results = pool.map(urllib2.urlopen, urls)

# Close the pool and wait for the work to finish
pool.close()
pool.join()

A wyniki pomiaru czasu:

Single thread:   14.4 seconds
       4 Pool:   3.1 seconds
       8 Pool:   1.4 seconds
      13 Pool:   1.3 seconds

Przekazywanie wielu argumentów (działa tak tylko w Pythonie 3.3 i nowszych ):

Aby przekazać wiele tablic:

results = pool.starmap(function, zip(list_a, list_b))

Lub przekazać stałą i tablicę:

results = pool.starmap(function, zip(itertools.repeat(constant), list_a))

Jeśli używasz wcześniejszej wersji Pythona, możesz przekazać wiele argumentów za pomocą tego obejścia ).

(Podziękowania dla user136036 za pomocny komentarz.)

philshem
źródło
90
Brakuje tylko głosów, ponieważ jest tak świeżo opublikowany. Ta odpowiedź działa pięknie i pokazuje funkcjonalność „mapy”, która zapewnia znacznie łatwiejszą do zrozumienia składnię niż inne odpowiedzi tutaj.
bezczynny
25
Czy to nawet wątki, a nie procesy? Wygląda na to, że próbuje się przetwarzać! = Wielowątkowy
AturSams,
72
Nawiasem mówiąc, możecie też pisać with Pool(8) as p: p.map( *whatever* )i pozbywać się linii księgowych.
11
@BarafuAlbino: Przydatne, bo warto chyba zauważyć, że działa to tylko w Pythonie 3.3+ .
fuglede
9
Jak pozostawić tę odpowiedź i nie wspominać, że jest to przydatne tylko w przypadku operacji we / wy? Działa to tylko na jednym wątku, który jest bezużyteczny w większości przypadków i jest w rzeczywistości wolniejszy niż zwykłe robienie tego
Frobot
714

Oto prosty przykład: musisz wypróbować kilka alternatywnych adresów URL i zwrócić zawartość pierwszego, aby odpowiedzieć.

import Queue
import threading
import urllib2

# Called by each thread
def get_url(q, url):
    q.put(urllib2.urlopen(url).read())

theurls = ["http://google.com", "http://yahoo.com"]

q = Queue.Queue()

for u in theurls:
    t = threading.Thread(target=get_url, args = (q,u))
    t.daemon = True
    t.start()

s = q.get()
print s

Jest to przypadek, w którym wątki są używane jako prosta optymalizacja: każdy wątek oczekuje na adres URL do rozwiązania i odpowiedzi, aby umieścić jego zawartość w kolejce; każdy wątek jest demonem (nie utrzyma procesu, jeśli główny wątek się skończy - jest to częstsze niż nie); główny wątek rozpoczyna wszystkie wątki, wykonuje getkolejkę, aby poczekać, aż jeden z nich wykona a put, a następnie emituje wyniki i kończy działanie (co powoduje usunięcie wszystkich wątków, które mogą być nadal uruchomione, ponieważ są one wątkami demonicznymi).

Prawidłowe użycie wątków w Pythonie jest niezmiennie połączone z operacjami We / Wy (ponieważ CPython i tak nie używa wielu rdzeni do uruchamiania zadań związanych z procesorem, jedynym powodem wątkowania nie jest blokowanie procesu podczas oczekiwania na niektóre operacje we / wy ). Nawiasem mówiąc, kolejki są prawie zawsze najlepszym sposobem na wyodrębnienie pracy do wątków i / lub zebranie wyników pracy, i są wewnętrznie bezpieczne dla wątków, dzięki czemu nie musisz się martwić o blokady, warunki, zdarzenia, semafory i inne inter -przeczytane koncepcje koordynacji / komunikacji.

Alex Martelli
źródło
10
Jeszcze raz dziękuję, MartelliBot. Zaktualizowałem przykład czekać na wszystko, aby odpowiedzieć adresów URL: import kolejce, gwintowania, urllib2 q = Queue.Queue () URL = '' ' a.com B.com . C.com ''' split () urls_received = 0 def get_url (q, url): req = urllib2.Request (url) resp = urllib2.urlopen (req) q.put (resp.read ()) global urls_received urls_received + = 1 print url_received for u in urls: t = Threading.Thread (target = get_url, args = (q, u)) t.daemon = True t.start () podczas gdy q.empty () i urls_received <len (urls): s = q.get () print s
htmldrum
3
@JRM: jeśli spojrzysz na następną odpowiedź poniżej, myślę, że lepszym sposobem oczekiwania na zakończenie wątków byłoby użycie tej join()metody, ponieważ sprawiłoby to, że główny wątek czekałby, aż zostaną wykonane bez ciągłego zużywania procesora sprawdzanie wartości. @Alex: dzięki, właśnie tego potrzebowałem, aby zrozumieć, jak używać wątków.
krs013
6
W przypadku python3 zamień „import urllib2” na „import urllib.request as urllib2”. i umieść nawiasy w instrukcji print.
Harvey
5
W przypadku python 3 zamień Queuenazwę modułu na queue. Nazwa metody jest taka sama.
JSmyth
2
Zwracam uwagę, że rozwiązanie wydrukuje tylko jedną ze stron. Aby wydrukować obie strony z kolejki, po prostu ponownie uruchom komendę: s = q.get() print s @ krs013 Nie potrzebujesz, joinponieważ Queue.get () blokuje.
Tom Anderson
256

UWAGA : Do faktycznej równoległości w Pythonie należy użyć modułu wieloprocesowego do rozwidlenia wielu procesów, które są wykonywane równolegle (ze względu na globalną blokadę interpretera wątki Pythona zapewniają przeplatanie, ale w rzeczywistości są wykonywane szeregowo, a nie równolegle i są tylko przydatne podczas przeplatania operacji we / wy).

Jeśli jednak szukasz jedynie przeplotu (lub wykonujesz operacje we / wy, które można zrównoleglić pomimo globalnej blokady interpretera), możesz zacząć od modułu wątków . Jako bardzo prosty przykład rozważmy problem sumowania dużego zakresu przez równoległe sumowanie podzakresów:

import threading

class SummingThread(threading.Thread):
     def __init__(self,low,high):
         super(SummingThread, self).__init__()
         self.low=low
         self.high=high
         self.total=0

     def run(self):
         for i in range(self.low,self.high):
             self.total+=i


thread1 = SummingThread(0,500000)
thread2 = SummingThread(500000,1000000)
thread1.start() # This actually causes the thread to run
thread2.start()
thread1.join()  # This waits until the thread has completed
thread2.join()
# At this point, both threads have completed
result = thread1.total + thread2.total
print result

Zauważ, że powyższy jest bardzo głupim przykładem, ponieważ absolutnie nie ma żadnych operacji we / wy i będzie wykonywany szeregowo, choć przeplatany (z dodatkowym narzutem przełączania kontekstu) w CPython z powodu globalnej blokady interpretera.

Michael Aaron Safyan
źródło
16
@Alex, nie powiedziałem, że to praktyczne, ale pokazuje, jak definiować i spawnować wątki, co moim zdaniem jest tym, czego chce OP.
Michael Aaron Safyan
6
Chociaż pokazuje to, jak definiować i odradzać wątki, w rzeczywistości nie sumuje podzakresów równolegle. thread1działa, dopóki nie zostanie zakończone, gdy główny wątek blokuje się, wtedy dzieje się to samo thread2, a następnie główny wątek wznawia się i drukuje zgromadzone wartości.
martineau
Czy nie powinno tak być super(SummingThread, self).__init__()? Jak w stackoverflow.com/a/2197625/806988
James Andres
@JamesAndres, zakładając, że nikt nie dziedziczy po „SummingThread”, wówczas jedno z nich działa dobrze; w takim przypadku super (SummingThread, self) jest tylko fantazyjnym sposobem na wyszukanie następnej klasy w kolejności rozdzielczości metody (MRO), która jest wątkiem. Gwint (a następnie wywołanie init w obu przypadkach). Masz jednak rację, ponieważ użycie super () jest lepszym stylem dla obecnego Pythona. Super było stosunkowo nowe w momencie, gdy podałem tę odpowiedź, dlatego też wywoływałem bezpośrednio do klasy super zamiast używać super (). Zaktualizuję to, aby użyć super.
Michael Aaron Safyan
14
OSTRZEŻENIE: Nie używaj wielowątkowości w takich zadaniach! Jak pokazał Dave Beazley: dabeaz.com/python/NewGIL.pdf , 2 wątki Pythona na 2 procesorach wykonują zadanie obciążające procesor 2 razy wolniej niż 1 wątek na 1 procesorze i 1,5 razy wolniej niż 2 wątki na 1 procesorze. To dziwne zachowanie wynika z niewłaściwej koordynacji wysiłków między systemem operacyjnym a Pythonem. Prawdziwy przypadek użycia wątków jest trudnym zadaniem we / wy. Np. Kiedy wykonujesz odczyt / zapis przez sieć, sensowne jest umieszczenie wątku, czekanie na odczyt / zapis danych, na tło i przełączenie procesora na inny wątek, który musi przetwarzać dane.
Boris Burkov
98

Jak inni wspomniano, CPython może używać wątków tylko do oczekiwań we / wy z powodu GIL .

Jeśli chcesz korzystać z wielu rdzeni dla zadań związanych z procesorem, użyj przetwarzania wieloprocesowego :

from multiprocessing import Process

def f(name):
    print 'hello', name

if __name__ == '__main__':
    p = Process(target=f, args=('bob',))
    p.start()
    p.join()
Kai
źródło
33
czy mógłbyś trochę wyjaśnić, co to robi?
pandita
5
@pandita: kod tworzy proces, a następnie go uruchamia. Tak więc teraz dzieją się dwie rzeczy: główna linia programu i proces rozpoczynający się od funkcji docelowej f. Równolegle program główny po prostu czeka na zakończenie procesu, co joinjest z nim zgodne. Jeśli główna część właśnie się zakończyła, podproces może, ale nie musi, działać do końca, więc robienie tego joinzawsze jest zalecane.
johntellsall
1
Rozszerzona odpowiedź obejmująca tę mapfunkcję znajduje się tutaj: stackoverflow.com/a/28463266/2327328
philshem
2
@philshem Uważać b / c linku pisał wykorzystuje pulę wątków (nie przetwarza) Jak wspomniano tutaj stackoverflow.com/questions/26432411/... . Jednak ta odpowiedź wykorzystuje proces. Jestem nowy w tym temacie, ale wydaje się, że (z powodu GIL) zyskasz na wydajności tylko w określonych sytuacjach, gdy korzystasz z wielowątkowości w Pythonie. Jednak korzystanie z puli procesów może korzystać z procesora wielordzeniowego, ponieważ proces ma więcej niż 1 rdzeń.
user3731622,
3
To najlepsza odpowiedź na to, aby rzeczywiście zrobić coś pożytecznego i skorzystać z wielu rdzeni procesora
Frobot
92

Tylko uwaga: kolejkowanie nie jest wymagane do tworzenia wątków.

To najprostszy przykład, jaki mogłem sobie wyobrazić, który pokazuje 10 procesów działających równolegle.

import threading
from random import randint
from time import sleep


def print_number(number):

    # Sleeps a random 1 to 10 seconds
    rand_int_var = randint(1, 10)
    sleep(rand_int_var)
    print "Thread " + str(number) + " slept for " + str(rand_int_var) + " seconds"

thread_list = []

for i in range(1, 10):

    # Instantiates the thread
    # (i) does not make a sequence, so (i,)
    t = threading.Thread(target=print_number, args=(i,))
    # Sticks the thread in a list so that it remains accessible
    thread_list.append(t)

# Starts threads
for thread in thread_list:
    thread.start()

# This blocks the calling thread until the thread whose join() method is called is terminated.
# From http://docs.python.org/2/library/threading.html#thread-objects
for thread in thread_list:
    thread.join()

# Demonstrates that the main process waited for threads to complete
print "Done"
Douglas Adams
źródło
3
Dodaj ostatni cytat do „Gotowe, aby wydrukować” „Gotowe”
iChux
1
Podoba mi się ten przykład bardziej niż Martelli, łatwiej się z nim bawić. Zaleciłbym jednak, aby printNumber wykonał następujące czynności, aby nieco wyjaśnić, co się dzieje: powinien zapisać randint do zmiennej przed spaniem na nim, a następnie wydruk powinien zostać zmieniony na „Thread” + str ( liczba) + „spał przez” + zmiennaandinta + „sekundy”
Nickolai,
Czy istnieje sposób, aby wiedzieć, kiedy każdy wątek się kończy, gdy się kończy?
Matt
1
@Matt Istnieje kilka sposobów na zrobienie czegoś takiego, ale zależy to od twoich potrzeb. Jednym ze sposobów byłoby zaktualizowanie singletona lub innej publicznie dostępnej zmiennej, która jest oglądana w pętli while i aktualizowana na końcu wątku.
Douglas Adams,
2
Nie potrzebujesz drugiej forpętli, możesz wywoływać thread.start()w pierwszej pętli.
Mark Mishyn
49

Odpowiedź Alexa Martellego pomogła mi. Jednak tutaj jest zmodyfikowana wersja, która moim zdaniem była bardziej przydatna (przynajmniej dla mnie).

Zaktualizowano: działa zarówno w Pythonie 2, jak i Pythonie 3

try:
    # For Python 3
    import queue
    from urllib.request import urlopen
except:
    # For Python 2 
    import Queue as queue
    from urllib2 import urlopen

import threading

worker_data = ['http://google.com', 'http://yahoo.com', 'http://bing.com']

# Load up a queue with your data. This will handle locking
q = queue.Queue()
for url in worker_data:
    q.put(url)

# Define a worker function
def worker(url_queue):
    queue_full = True
    while queue_full:
        try:
            # Get your data off the queue, and do some work
            url = url_queue.get(False)
            data = urlopen(url).read()
            print(len(data))

        except queue.Empty:
            queue_full = False

# Create as many threads as you want
thread_count = 5
for i in range(thread_count):
    t = threading.Thread(target=worker, args = (q,))
    t.start()
JimJty
źródło
6
Dlaczego nie złamać wyjątku?
Stavros Korokithakis
1
możesz, po prostu osobiste preferencje
JimJty
1
Nie uruchomiłem kodu, ale czy nie musisz demonizować wątków? Myślę, że po ostatniej pętli for twój program może wyjść - przynajmniej powinien, bo tak powinny działać wątki. Myślę, że lepszym podejściem nie jest umieszczanie danych roboczych w kolejce, ale umieszczanie danych wyjściowych w kolejce, ponieważ wtedy można mieć pętlę główną, która nie tylko obsługuje informacje przychodzące do kolejki od pracowników, ale teraz również nie wątkuje, i wiesz, że nie wyjdzie przedwcześnie.
dylnmc
1
@dylnmc, to jest poza moim przypadkiem użycia (moja kolejka wejściowa jest predefiniowana). Jeśli chcesz iść swoją trasą, proponuję spojrzeć na selera
JimJty
@JimJty, czy wiesz, dlaczego pojawia się ten błąd: import Queue ModuleNotFoundError: No module named 'Queue'działam w Pythonie 3.6.5 niektóre posty wspominają, że w Pythonie 3.6.5 jest, queueale nawet po jego zmianie nadal nie działa
9371654
25

Biorąc pod uwagę funkcję f, napisz to w ten sposób:

import threading
threading.Thread(target=f).start()

Aby przekazać argumenty do f

threading.Thread(target=f, args=(a,b,c)).start()
rozgwiazdy
źródło
To jest bardzo proste. W jaki sposób upewniasz się, że wątki zamykają się, gdy z nimi skończysz?
cameronroytaylor
O ile rozumiem, po wyjściu z funkcji Threadobiekt oczyszcza się. Zobacz dokumenty . Istnieje is_alive()metoda, której można użyć do sprawdzenia wątku, jeśli zajdzie taka potrzeba.
starfry
Widziałem tę is_alivemetodę, ale nie mogłem wymyślić, jak zastosować ją do wątku. Próbowałem przypisać, thread1=threading.Thread(target=f).start()a następnie sprawdzić to thread1.is_alive(), ale thread1jest wypełnione None, więc nie ma szczęścia. Czy wiesz, czy istnieje inny sposób uzyskania dostępu do wątku?
cameronroytaylor
4
Musisz przypisać obiekt wątku do zmiennej, a następnie uruchomić go za pomocą tej zmiennej: thread1=threading.Thread(target=f)następnie thread1.start(). To możesz zrobić thread1.is_alive().
starfry
1
To się udało. I tak, testowanie ze thread1.is_alive()zwrotami, Falsejak tylko funkcja wyjdzie.
cameronroytaylor
25

Uznałem to za bardzo przydatne: utwórz tyle wątków, ile rdzeni i pozwól im wykonywać (dużą) liczbę zadań (w tym przypadku wywoływanie programu powłoki):

import Queue
import threading
import multiprocessing
import subprocess

q = Queue.Queue()
for i in range(30): # Put 30 tasks in the queue
    q.put(i)

def worker():
    while True:
        item = q.get()
        # Execute a task: call a shell program and wait until it completes
        subprocess.call("echo " + str(item), shell=True)
        q.task_done()

cpus = multiprocessing.cpu_count() # Detect number of cores
print("Creating %d threads" % cpus)
for i in range(cpus):
     t = threading.Thread(target=worker)
     t.daemon = True
     t.start()

q.join() # Block until all tasks are done
delfin
źródło
@shavenwarthog na pewno można dostosować zmienną „cpus” w zależności od potrzeb. W każdym razie wywołanie podprocesu spowoduje odrodzenie podprocesów, które zostaną przydzielone przez system operacyjny (proces „nadrzędny” pytona nie oznacza „tego samego procesora” dla podprocesów).
delfin
2
masz rację, mój komentarz na temat „wątków jest uruchamianych na tym samym procesorze co proces nadrzędny” jest błędny. Dziękuję za odpowiedź!
johntellsall
1
być może warto zauważyć, że w przeciwieństwie do wielowątkowości, która wykorzystuje tę samą przestrzeń pamięci, wieloprocesowość nie może tak łatwo udostępniać zmiennych / danych. +1.
fantastyczny
22

Python 3 ma możliwość uruchamiania równoległych zadań . Ułatwia to naszą pracę.

Ma pule wątków i pule procesów .

Poniżej przedstawiono wgląd:

ThreadPoolExecutor Przykład ( źródło )

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/']

# Retrieve a single page and report the URL and contents
def load_url(url, timeout):
    with urllib.request.urlopen(url, timeout=timeout) as conn:
        return conn.read()

# We can use a with statement to ensure threads are cleaned up promptly
with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
    # Start the load operations and mark each future with its URL
    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)))

ProcessPoolExecutor ( źródło )

import concurrent.futures
import math

PRIMES = [
    112272535095293,
    112582705942171,
    112272535095293,
    115280095190773,
    115797848077099,
    1099726899285419]

def is_prime(n):
    if n % 2 == 0:
        return False

    sqrt_n = int(math.floor(math.sqrt(n)))
    for i in range(3, sqrt_n + 1, 2):
        if n % i == 0:
            return False
    return True

def main():
    with concurrent.futures.ProcessPoolExecutor() as executor:
        for number, prime in zip(PRIMES, executor.map(is_prime, PRIMES)):
            print('%d is prime: %s' % (number, prime))

if __name__ == '__main__':
    main()
Jeril
źródło
18

Korzystanie z nowego modułu concurrent.futures

def sqr(val):
    import time
    time.sleep(0.1)
    return val * val

def process_result(result):
    print(result)

def process_these_asap(tasks):
    import concurrent.futures

    with concurrent.futures.ProcessPoolExecutor() as executor:
        futures = []
        for task in tasks:
            futures.append(executor.submit(sqr, task))

        for future in concurrent.futures.as_completed(futures):
            process_result(future.result())
        # Or instead of all this just do:
        # results = executor.map(sqr, tasks)
        # list(map(process_result, results))

def main():
    tasks = list(range(10))
    print('Processing {} tasks'.format(len(tasks)))
    process_these_asap(tasks)
    print('Done')
    return 0

if __name__ == '__main__':
    import sys
    sys.exit(main())

Podejście do executorów może wydawać się znajome dla wszystkich, którzy wcześniej brudzili sobie ręce Javą.

Na marginesie: Aby zachować rozsądek we wszechświecie, nie zapomnij zamknąć puli / executorów, jeśli nie używasz withkontekstu (co jest tak niesamowite, że robi to za Ciebie)

Shubham Chaudhary
źródło
17

Dla mnie doskonałym przykładem wątków jest monitorowanie zdarzeń asynchronicznych. Spójrz na ten kod.

# thread_test.py
import threading
import time

class Monitor(threading.Thread):
    def __init__(self, mon):
        threading.Thread.__init__(self)
        self.mon = mon

    def run(self):
        while True:
            if self.mon[0] == 2:
                print "Mon = 2"
                self.mon[0] = 3;

Możesz grać z tym kodem, otwierając sesję IPython i robiąc coś takiego:

>>> from thread_test import Monitor
>>> a = [0]
>>> mon = Monitor(a)
>>> mon.start()
>>> a[0] = 2
Mon = 2
>>>a[0] = 2
Mon = 2

Zaczekaj kilka minut

>>> a[0] = 2
Mon = 2
dvreed77
źródło
1
AttributeError: Obiekt „Monitor” nie ma atrybutu „stop”?
pandita
5
Czy nie odpalasz cykli procesora podczas oczekiwania na wydarzenie? Nie zawsze bardzo praktyczna rzecz do zrobienia.
potentat
3
Jak mówi potentat, będzie to stale wykonywane. Przynajmniej możesz dodać w krótkim śnie, powiedzmy sen (0.1), co prawdopodobnie znacznie zmniejszy użycie procesora na prostym przykładzie takim jak ten.
fantastyczny
3
To okropny przykład marnowania jednego rdzenia. Dodaj przynajmniej sen, ale właściwym rozwiązaniem jest użycie mechanizmu sygnalizacyjnego.
PureW,
16

Większość dokumentacji i samouczków używa Pythona Threadingi Queuemodułu i mogą one wydawać się przytłaczające dla początkujących.

Być może rozważ concurrent.futures.ThreadPoolExecutormoduł Python 3.

W połączeniu z withklauzulami i listami może to być prawdziwy urok.

from concurrent.futures import ThreadPoolExecutor, as_completed

def get_url(url):
    # Your actual program here. Using threading.Lock() if necessary
    return ""

# List of URLs to fetch
urls = ["url1", "url2"]

with ThreadPoolExecutor(max_workers = 5) as executor:

    # Create threads
    futures = {executor.submit(get_url, url) for url in urls}

    # as_completed() gives you the threads once finished
    for f in as_completed(futures):
        # Get the results
        rs = f.result()
Yibo
źródło
15

Widziałem tutaj wiele przykładów, w których nie wykonywano żadnej prawdziwej pracy, i były one głównie związane z procesorem. Oto przykład zadania związanego z procesorem, które oblicza wszystkie liczby pierwsze między 10 milionów a 10,05 miliona. Użyłem tutaj wszystkich czterech metod:

import math
import timeit
import threading
import multiprocessing
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor


def time_stuff(fn):
    """
    Measure time of execution of a function
    """
    def wrapper(*args, **kwargs):
        t0 = timeit.default_timer()
        fn(*args, **kwargs)
        t1 = timeit.default_timer()
        print("{} seconds".format(t1 - t0))
    return wrapper

def find_primes_in(nmin, nmax):
    """
    Compute a list of prime numbers between the given minimum and maximum arguments
    """
    primes = []

    # Loop from minimum to maximum
    for current in range(nmin, nmax + 1):

        # Take the square root of the current number
        sqrt_n = int(math.sqrt(current))
        found = False

        # Check if the any number from 2 to the square root + 1 divides the current numnber under consideration
        for number in range(2, sqrt_n + 1):

            # If divisible we have found a factor, hence this is not a prime number, lets move to the next one
            if current % number == 0:
                found = True
                break

        # If not divisible, add this number to the list of primes that we have found so far
        if not found:
            primes.append(current)

    # I am merely printing the length of the array containing all the primes, but feel free to do what you want
    print(len(primes))

@time_stuff
def sequential_prime_finder(nmin, nmax):
    """
    Use the main process and main thread to compute everything in this case
    """
    find_primes_in(nmin, nmax)

@time_stuff
def threading_prime_finder(nmin, nmax):
    """
    If the minimum is 1000 and the maximum is 2000 and we have four workers,
    1000 - 1250 to worker 1
    1250 - 1500 to worker 2
    1500 - 1750 to worker 3
    1750 - 2000 to worker 4
    so let’s split the minimum and maximum values according to the number of workers
    """
    nrange = nmax - nmin
    threads = []
    for i in range(8):
        start = int(nmin + i * nrange/8)
        end = int(nmin + (i + 1) * nrange/8)

        # Start the thread with the minimum and maximum split up to compute
        # Parallel computation will not work here due to the GIL since this is a CPU-bound task
        t = threading.Thread(target = find_primes_in, args = (start, end))
        threads.append(t)
        t.start()

    # Don’t forget to wait for the threads to finish
    for t in threads:
        t.join()

@time_stuff
def processing_prime_finder(nmin, nmax):
    """
    Split the minimum, maximum interval similar to the threading method above, but use processes this time
    """
    nrange = nmax - nmin
    processes = []
    for i in range(8):
        start = int(nmin + i * nrange/8)
        end = int(nmin + (i + 1) * nrange/8)
        p = multiprocessing.Process(target = find_primes_in, args = (start, end))
        processes.append(p)
        p.start()

    for p in processes:
        p.join()

@time_stuff
def thread_executor_prime_finder(nmin, nmax):
    """
    Split the min max interval similar to the threading method, but use a thread pool executor this time.
    This method is slightly faster than using pure threading as the pools manage threads more efficiently.
    This method is still slow due to the GIL limitations since we are doing a CPU-bound task.
    """
    nrange = nmax - nmin
    with ThreadPoolExecutor(max_workers = 8) as e:
        for i in range(8):
            start = int(nmin + i * nrange/8)
            end = int(nmin + (i + 1) * nrange/8)
            e.submit(find_primes_in, start, end)

@time_stuff
def process_executor_prime_finder(nmin, nmax):
    """
    Split the min max interval similar to the threading method, but use the process pool executor.
    This is the fastest method recorded so far as it manages process efficiently + overcomes GIL limitations.
    RECOMMENDED METHOD FOR CPU-BOUND TASKS
    """
    nrange = nmax - nmin
    with ProcessPoolExecutor(max_workers = 8) as e:
        for i in range(8):
            start = int(nmin + i * nrange/8)
            end = int(nmin + (i + 1) * nrange/8)
            e.submit(find_primes_in, start, end)

def main():
    nmin = int(1e7)
    nmax = int(1.05e7)
    print("Sequential Prime Finder Starting")
    sequential_prime_finder(nmin, nmax)
    print("Threading Prime Finder Starting")
    threading_prime_finder(nmin, nmax)
    print("Processing Prime Finder Starting")
    processing_prime_finder(nmin, nmax)
    print("Thread Executor Prime Finder Starting")
    thread_executor_prime_finder(nmin, nmax)
    print("Process Executor Finder Starting")
    process_executor_prime_finder(nmin, nmax)

main()

Oto wyniki na moim czterordzeniowym komputerze Mac OS X.

Sequential Prime Finder Starting
9.708213827005238 seconds
Threading Prime Finder Starting
9.81836523200036 seconds
Processing Prime Finder Starting
3.2467174359990167 seconds
Thread Executor Prime Finder Starting
10.228896902000997 seconds
Process Executor Finder Starting
2.656402041000547 seconds
PirateApp
źródło
1
@TheUnfunCat żaden wykonawca procesu nie jest o wiele lepszy niż wątkowanie zadań związanych z procesorem
PirateApp
1
Świetna odpowiedź koleś. Mogę potwierdzić, że w Pythonie 3.6 w systemie Windows (przynajmniej) ThreadPoolExecutor nie robi nic dobrego dla zadań obciążających procesor. Nie wykorzystuje rdzeni do obliczeń. Podczas gdy ProcessPoolExecutor kopiuje dane do KAŻDEGO procesu, który się odradza, jest zabójczy dla dużych matryc.
Anatolij Aleksiejew
1
Bardzo przydatny przykład, ale nie rozumiem, jak to kiedykolwiek działało. Potrzebujemy if __name__ == '__main__':przed głównym wezwanie, w przeciwnym razie ikra siebie i drukuje pomiarowe Próba zostało wykonane, aby rozpocząć nowy proces przed ... .
Stein
1
@ Stein Uważam, że to tylko problem w systemie Windows.
AMC
12

Oto bardzo prosty przykład importu CSV za pomocą wątków. (Włączenie biblioteki może się różnić w zależności od celu).

Funkcje pomocnicze:

from threading import Thread
from project import app
import csv


def import_handler(csv_file_name):
    thr = Thread(target=dump_async_csv_data, args=[csv_file_name])
    thr.start()

def dump_async_csv_data(csv_file_name):
    with app.app_context():
        with open(csv_file_name) as File:
            reader = csv.DictReader(File)
            for row in reader:
                # DB operation/query

Funkcja kierowcy:

import_handler(csv_file_name)
Chirag Vora
źródło
9

Chciałbym przedstawić prosty przykład i wyjaśnienia, które okazały się przydatne, gdy sam musiałem rozwiązać ten problem.

W tej odpowiedzi znajdziesz informacje o GIL Pythona (globalna blokada interpretera) i prosty przykład z dnia na dzień napisany przy użyciu multiprocessing.dummy oraz kilka prostych testów porównawczych.

Global Interpreter Lock (GIL)

Python nie pozwala na wielowątkowość w najprawdziwszym tego słowa znaczeniu. Ma pakiet wielowątkowy, ale jeśli chcesz wielowątkowy, aby przyspieszyć swój kod, zwykle nie jest dobrym pomysłem, aby go użyć.

Python ma konstrukcję zwaną globalną blokadą interpretera (GIL). GIL zapewnia, że ​​tylko jeden z twoich „wątków” może być wykonywany jednocześnie. Wątek nabywa GIL, wykonuje niewielką pracę, a następnie przekazuje GIL do następnego wątku.

Dzieje się to bardzo szybko, więc dla ludzkiego oka może się wydawać, że twoje wątki wykonują się równolegle, ale tak naprawdę po prostu korzystają z tego samego rdzenia procesora.

Całe to przekazywanie GIL zwiększa koszty wykonania. Oznacza to, że jeśli chcesz, aby Twój kod działał szybciej, korzystanie z pakietu wątków często nie jest dobrym pomysłem.

Istnieją powody, aby używać pakietu wątków Pythona. Jeśli chcesz uruchomić kilka rzeczy jednocześnie, a wydajność nie jest problemem, to jest całkowicie w porządku i wygodne. Lub jeśli korzystasz z kodu, który musi na coś czekać (np. Niektóre wejścia / wyjścia), może to mieć sens. Ale biblioteka wątków nie pozwala na użycie dodatkowych rdzeni procesora.

Wielowątkowość można zlecić systemowi operacyjnemu (poprzez przetwarzanie wieloskładnikowe) i niektórym zewnętrznym aplikacjom, które wywołują Twój kod Python (na przykład Spark lub Hadoop ), lub niektóre kody, które wywołuje Twój kod Python (na przykład: możesz twój kod Python wywołuje funkcję C, która wykonuje drogie wielowątkowe rzeczy).

Dlaczego to ma znaczenie

Ponieważ wiele osób spędza dużo czasu próbując znaleźć wąskie gardła w swoim fantazyjnym wielowątkowym kodzie Pythona, zanim dowiedzą się, czym jest GIL.

Gdy te informacje będą jasne, oto mój kod:

#!/bin/python
from multiprocessing.dummy import Pool
from subprocess import PIPE,Popen
import time
import os

# In the variable pool_size we define the "parallelness".
# For CPU-bound tasks, it doesn't make sense to create more Pool processes
# than you have cores to run them on.
#
# On the other hand, if you are using I/O-bound tasks, it may make sense
# to create a quite a few more Pool processes than cores, since the processes
# will probably spend most their time blocked (waiting for I/O to complete).
pool_size = 8

def do_ping(ip):
    if os.name == 'nt':
        print ("Using Windows Ping to " + ip)
        proc = Popen(['ping', ip], stdout=PIPE)
        return proc.communicate()[0]
    else:
        print ("Using Linux / Unix Ping to " + ip)
        proc = Popen(['ping', ip, '-c', '4'], stdout=PIPE)
        return proc.communicate()[0]


os.system('cls' if os.name=='nt' else 'clear')
print ("Running using threads\n")
start_time = time.time()
pool = Pool(pool_size)
website_names = ["www.google.com","www.facebook.com","www.pinterest.com","www.microsoft.com"]
result = {}
for website_name in website_names:
    result[website_name] = pool.apply_async(do_ping, args=(website_name,))
pool.close()
pool.join()
print ("\n--- Execution took {} seconds ---".format((time.time() - start_time)))

# Now we do the same without threading, just to compare time
print ("\nRunning NOT using threads\n")
start_time = time.time()
for website_name in website_names:
    do_ping(website_name)
print ("\n--- Execution took {} seconds ---".format((time.time() - start_time)))

# Here's one way to print the final output from the threads
output = {}
for key, value in result.items():
    output[key] = value.get()
print ("\nOutput aggregated in a Dictionary:")
print (output)
print ("\n")

print ("\nPretty printed output: ")
for key, value in output.items():
    print (key + "\n")
    print (value)
Pitto
źródło
7

Oto wielowątkowość z prostym przykładem, który będzie pomocny. Możesz go uruchomić i łatwo zrozumieć, jak działa wielowątkowość w Pythonie. Użyłem blokady, aby uniemożliwić dostęp do innych wątków, dopóki poprzednie wątki nie zakończyły swojej pracy. Korzystając z tego wiersza kodu,

tLock = Threading.BoundedSemaphore (wartość = 4)

możesz zezwolić na wiele procesów na raz i trzymać resztę wątków, które będą uruchamiane później lub po zakończeniu poprzednich procesów.

import threading
import time

#tLock = threading.Lock()
tLock = threading.BoundedSemaphore(value=4)
def timer(name, delay, repeat):
    print  "\r\nTimer: ", name, " Started"
    tLock.acquire()
    print "\r\n", name, " has the acquired the lock"
    while repeat > 0:
        time.sleep(delay)
        print "\r\n", name, ": ", str(time.ctime(time.time()))
        repeat -= 1

    print "\r\n", name, " is releaseing the lock"
    tLock.release()
    print "\r\nTimer: ", name, " Completed"

def Main():
    t1 = threading.Thread(target=timer, args=("Timer1", 2, 5))
    t2 = threading.Thread(target=timer, args=("Timer2", 3, 5))
    t3 = threading.Thread(target=timer, args=("Timer3", 4, 5))
    t4 = threading.Thread(target=timer, args=("Timer4", 5, 5))
    t5 = threading.Thread(target=timer, args=("Timer5", 0.1, 5))

    t1.start()
    t2.start()
    t3.start()
    t4.start()
    t5.start()

    print "\r\nMain Complete"

if __name__ == "__main__":
    Main()
cSharma
źródło
5

Pożyczając z tego postu wiemy o wyborze między wielowątkowością, wieloprocesowością i asynchronią / asyncioi ich wykorzystaniem.

Python 3 ma nową wbudowaną bibliotekę w celu współbieżności i równoległości: concurrent.futures

Pokażę więc poprzez eksperyment, aby uruchomić cztery zadania (tj. .sleep()Metodę) według Threading-Poolsposobu:

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

def concurrent(max_worker=1):
    futures = []

    tick = time()
    with ThreadPoolExecutor(max_workers=max_worker) as executor:
        futures.append(executor.submit(sleep, 2))  # Two seconds sleep
        futures.append(executor.submit(sleep, 1))
        futures.append(executor.submit(sleep, 7))
        futures.append(executor.submit(sleep, 3))

        for future in as_completed(futures):
            if future.result() is not None:
                print(future.result())

    print('Total elapsed time by {} workers:'.format(max_worker), time()-tick)

concurrent(5)
concurrent(4)
concurrent(3)
concurrent(2)
concurrent(1)

Wynik:

Total elapsed time by 5 workers: 7.007831811904907
Total elapsed time by 4 workers: 7.007944107055664
Total elapsed time by 3 workers: 7.003149509429932
Total elapsed time by 2 workers: 8.004627466201782
Total elapsed time by 1 workers: 13.013478994369507

[ UWAGA ]:

  • Jak widać na powyższych wynikach, najlepszym przypadkiem było 3 pracowników dla tych czterech zadań.
  • Jeśli masz zadanie procesu zamiast I / O związany lub blokowanie ( multiprocessingvs threading) można zmienić ThreadPoolExecutorna ProcessPoolExecutor.
Benyamin Jafari
źródło
4

Żadne z poprzednich rozwiązań nie używało wielu rdzeni na moim serwerze GNU / Linux (gdzie nie mam uprawnień administratora). Po prostu działały na jednym rdzeniu.

Użyłem os.forkinterfejsu niższego poziomu do odrodzenia wielu procesów. Oto kod, który zadziałał dla mnie:

from os import fork

values = ['different', 'values', 'for', 'threads']

for i in range(len(values)):
    p = fork()
    if p == 0:
        my_function(values[i])
        break
David Schumann
źródło
2
import threading
import requests

def send():

  r = requests.get('https://www.stackoverlow.com')

thread = []
t = threading.Thread(target=send())
thread.append(t)
t.start()
Skiller Dz
źródło
1
@sP_ Zgaduję, bo wtedy masz obiekty wątku, więc możesz poczekać, aż się skończą.
Aleksandar Makragić
1
t = gwintowanie. wątek (cel = wysyłanie ()) powinien być t = gwintowanie.
wątek
Głosuję za odrzuceniem tej odpowiedzi, ponieważ nie zawiera ona wyjaśnienia, w jaki sposób poprawia ona istniejące odpowiedzi, poza tym, że zawiera poważne niedokładności.
Jules