Bardzo proste wielowątkowe pobieranie równoległego adresu URL (bez kolejki)

Question

Spędziłem cały dzień szukając najprostszego możliwego pobierania znacznika URL w Pythonie, ale większość skryptów, które znalazłem, używa kolejek lub wieloprocesowych lub złożonych bibliotek.

Nareszcie napisałem jedną, którą zgłaszam jako odpowiedź. Proszę zasugerować jakąkolwiek poprawę.

Sądzę, że inne osoby mogły szukać czegoś podobnego.

Answer 1

Uproszczenie oryginalną wersję miarę możliwości:

import threading 
import urllib2 
import time 

start = time.time() 
urls = ["http://www.google.com", "http://www.apple.com", "http://www.microsoft.com", "http://www.amazon.com", "http://www.facebook.com"] 

def fetch_url(url): 
    urlHandler = urllib2.urlopen(url) 
    html = urlHandler.read() 
    print "'%s\' fetched in %ss" % (url, (time.time() - start)) 

threads = [threading.Thread(target=fetch_url, args=(url,)) for url in urls] 
for thread in threads: 
    thread.start() 
for thread in threads: 
    thread.join() 

print "Elapsed Time: %s" % (time.time() - start) 

Jedyne nowe sztuczki są tu:

• Śledzić wątków tworzonych.
• Nie przejmuj się licznikiem wątków, jeśli chcesz wiedzieć, kiedy wszystko jest gotowe; join już ci to mówi.
• Jeśli nie potrzebujesz żadnego stanu lub zewnętrznego interfejsu API, nie potrzebujesz podklasy Thread, tylko funkcji target.

Answer 2

Ten skrypt pobiera zawartość z zestawu adresów URL zdefiniowanych w tablicy. Tworzy wątek dla każdego adresu URL, który ma być pobrany, więc jest przeznaczony do użycia w ograniczonym zestawie adresów URL.

Zamiast używać obiektu kolejki, każdy wątek powiadamia swój koniec o wywołaniu zwrotnym do funkcji globalnej, która utrzymuje liczbę uruchomionych wątków.

import threading 
import urllib2 
import time 

start = time.time() 
urls = ["http://www.google.com", "http://www.apple.com", "http://www.microsoft.com", "http://www.amazon.com", "http://www.facebook.com"] 
left_to_fetch = len(urls) 

class FetchUrl(threading.Thread): 
    def __init__(self, url): 
     threading.Thread.__init__(self) 
     self.setDaemon = True 
     self.url = url 

    def run(self): 
     urlHandler = urllib2.urlopen(self.url) 
     html = urlHandler.read() 
     finished_fetch_url(self.url) 


def finished_fetch_url(url): 
    "callback function called when a FetchUrl thread ends" 
    print "\"%s\" fetched in %ss" % (url,(time.time() - start)) 
    global left_to_fetch 
    left_to_fetch-=1 
    if left_to_fetch==0: 
     "all urls have been fetched" 
     print "Elapsed Time: %ss" % (time.time() - start) 


for url in urls: 
    "spawning a FetchUrl thread for each url to fetch" 
    FetchUrl(url).start() 

Answer 3

Główny przykład w concurrent.futures robi wszystko, co chcesz, o wiele prostszy. Co więcej, może obsłużyć ogromną liczbę adresów URL, robiąc tylko 5 jednocześnie i znacznie lepiej radzi sobie z błędami.

Oczywiście ten moduł jest zbudowany tylko z Pythonem 3.2 lub nowszym ... ale jeśli używasz 2.5-3.1, możesz po prostu zainstalować backport, futures, z PyPI. Aby zmienić kod przykładowy, wystarczy wyszukać i zastąpić concurrent.futures z futures, a dla 2.x, urllib.request z urllib2.

Oto próbka przeniesiona do 2.x, zmodyfikowane do wykorzystania lista adresów URL i dodać razy:

import concurrent.futures 
import urllib2 
import time 

start = time.time() 
urls = ["http://www.google.com", "http://www.apple.com", "http://www.microsoft.com", "http://www.amazon.com", "http://www.facebook.com"] 

# Retrieve a single page and report the url and contents 
def load_url(url, timeout): 
    conn = urllib2.urlopen(url, timeout=timeout) 
    return conn.readall() 

# 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 '"%s" fetched in %ss' % (url,(time.time() - start)) 
print "Elapsed Time: %ss" % (time.time() - start) 

---

Ale można zrobić to jeszcze prostsze. Naprawdę, wszystko czego potrzebujesz to:

def load_url(url): 
    conn = urllib2.urlopen(url, timeout) 
    data = conn.readall() 
    print '"%s" fetched in %ss' % (url,(time.time() - start)) 
    return data 

with futures.ThreadPoolExecutor(max_workers=5) as executor: 
    pages = executor.map(load_url, urls) 

print "Elapsed Time: %ss" % (time.time() - start) 

Answer 4

Jestem teraz publikowania innego rozwiązania, przez posiadające gwinty pracownik nie-demona i łączenia ich do głównego wątku (co oznacza zablokowanie głównego wątku, dopóki wszystkich wątków roboczych zakończono) zamiast powiadamiać o zakończeniu wykonywania każdego wątku roboczego z wywołaniem zwrotnym do funkcji globalnej (tak jak to zrobiłem w poprzedniej odpowiedzi), ponieważ w niektórych komentarzach zauważono, że taki sposób nie jest bezpieczny dla wątków.

import threading 
import urllib2 
import time 

start = time.time() 
urls = ["http://www.google.com", "http://www.apple.com", "http://www.microsoft.com", "http://www.amazon.com", "http://www.facebook.com"] 

class FetchUrl(threading.Thread): 
    def __init__(self, url): 
     threading.Thread.__init__(self) 
     self.url = url 

    def run(self): 
     urlHandler = urllib2.urlopen(self.url) 
     html = urlHandler.read() 
     print "'%s\' fetched in %ss" % (self.url,(time.time() - start)) 

for url in urls: 
    FetchUrl(url).start() 

#Join all existing threads to main thread. 
for thread in threading.enumerate(): 
    if thread is not threading.currentThread(): 
     thread.join() 

print "Elapsed Time: %s" % (time.time() - start) 

Answer 5

multiprocessing posiada basen gwintu, który nie jest uruchamiany inne procesy:

#!/usr/bin/env python 
from multiprocessing.pool import ThreadPool 
from time import time as timer 
from urllib2 import urlopen 

urls = ["http://www.google.com", "http://www.apple.com", "http://www.microsoft.com", "http://www.amazon.com", "http://www.facebook.com"] 

def fetch_url(url): 
    try: 
     response = urlopen(url) 
     return url, response.read(), None 
    except Exception as e: 
     return url, None, e 

start = timer() 
results = ThreadPool(20).imap_unordered(fetch_url, urls) 
for url, html, error in results: 
    if error is None: 
     print("%r fetched in %ss" % (url, timer() - start)) 
    else: 
     print("error fetching %r: %s" % (url, error)) 
print("Elapsed Time: %s" % (timer() - start,)) 

Zalety w porównaniu z roztworem opartym Thread:

• ThreadPool pozwala ograniczyć maksymalną liczbę jednoczesnych połączenia (20 w kodzie przykładowym)
• wyjście nie jest zniekształcone, ponieważ wszystkie wyjścia są w e główny wątek
błędy są rejestrowane
• kod działa zarówno na Pythonie 2 i 3 bez zmian (przyjmując from urllib.request import urlopen na Pythonie 3).