многопоточная python программа не выполняет многопоточную функцию

Question

Я не имел опыта работы с потоками в python и пытался сделать несколько простых многопоточных программ, чтобы получить немного больше опыта. Я пытаюсь отправить запросы на заранее определенный список URL-адресов.

При попытке выполнить программу она мгновенно завершает работу и печатает («Конец») без выходов или исключений. вызов печати, помещенный в thread_function, не выполняется, и никаких ошибок не возникает.

Любая помощь приветствуется.

import networking
import threading
import concurrent.futures

class concurrencyTest:
    def __init__(self, URLlist):
        self.URLlist = URLlist
        self.resourceDict = {}


        self._urlListLock = threading.Lock()
        self._resourceListLock = threading.Lock()

    def sendMultiThreadedRequests(self, threadNum=3):
        self.resourceDict = {}
        with concurrent.futures.ThreadPoolExecutor(max_workers=threadNum) as executor:
            results = executor.map(self.thread_function)
        

    
    def thread_function(self):
        print("You are are in the thread_function")
        while True:
            with self._urlListLock:
                numOfRemainingURL = len(self.URLlist)
                print(numOfRemainingURL)
                if numOfRemainingURL == 0:
                    return

                urlToRequest = self.URLlist.pop()

            webpage = networking.getWebpage(urlToRequest)
            ##parse webpage or resource
            
            with self._resourceListLock:
                self.resourceDict[urlToRequest] = webpage
   
    
    def sendRegularRequests(self):
        self.resourceDict = {}
        for url in self.URLlist:
            resource = networking.getWebpage(url)
            self.resourceDict[url] = resource

    def updateURLpool(self):
        return "Not currently coded"


                  
def main():
    #The real urlList is a lot larger than just 3 URLs
    urlList = ["www.google.com","www.stackoverflow.com","www.reddit.com"]

    parTest = concurrencyTest(urlList)

    parTest.sendMultiThreadedRequests()
    
    print("End")

main()

Answer 1

executor.map() предназначен для сопоставления списка значений с вызовом функции и ожидает итерацию (например, список) в качестве второго аргумента (или количество объектов в качестве независимых аргументов) для сопоставления его содержимого с функцией, предоставленной как первый аргумент.

Например:

executor.map(self.thread_function, self.URLlist)

или

executor.map(self.thread_function, url1, url2, url3, ..., urln)

вызовет thread_function(url) для каждого значения в URLlist или каждого аргумента, предоставленного во втором пример.

Это, в свою очередь, означает, что ваша функция thread_function() должна принять аргумент, чтобы получить значение из списка: thread_function(self, url). Поскольку функция теперь получает только одно значение URLlist за раз, while l oop в вашей функции больше не имеет смысла, и вам нужно реорганизовать эту функцию для обработки только одного URL-адреса вместо списка:

def thread_function(self, url):
    webpage = getWebpage(url)

    # parse webpage or resource
        
    with self._resourceListLock:
        self.resourceDict[url] = webpage

В качестве альтернативы вы можете использовать submit() вместо map(), цель которого состоит в том, чтобы просто выполнить функцию асинхронно. Таким образом, модификация thread_function() не требуется:

executor.submit(self.thread_function)

Answer 2

Если вы хотите использовать concurrent.futures

, вы никогда не передаете никаких итераций в .map(), поэтому ничего не будет сделано. Чтобы упростить то, что у вас есть (вам также не нужны никакие блокировки):

import concurrent.futures
import random
import time
import hashlib


def get_data(url):
    print(f"Starting to get {url}")
    # to pretend doing some work:
    time.sleep(random.uniform(0.5, 1))
    result = hashlib.sha1(url.encode("utf-8")).hexdigest()  
    print(f"OK: {url}")
    return (url, result)


url_list = ["www.google.com", "www.stackoverflow.com", "www.reddit.com"]
with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
    results = {}
    for key, value in executor.map(get_data, url_list):
        results[key] = value
        print(f"Results acquired: {len(results)}")
    # or more simply
    # results = dict(executor.map(get_data, url_list))
    print(results)

распечатывает (например, случайный)

Starting to get www.google.com
Starting to get www.stackoverflow.com
Starting to get www.reddit.com
OK: www.google.com
Results acquired: 1
OK: www.stackoverflow.com
Results acquired: 2
OK: www.reddit.com
Results acquired: 3
{'www.google.com': 'd8b99f68b208b5453b391cb0c6c3d6a9824f3c3a', 'www.stackoverflow.com': '3954ca3139369180fff4ea3ae984b9a7871b540d', 'www.reddit.com': 'f420470addba27b8577bb40e02229e90af568d69'}

Если вы хотите использовать multiprocessing

(та же функция get_data, что и выше)

from multiprocessing.pool import ThreadPool, Pool
# (choose between threads or processes)

with ThreadPool(3) as p:
    results = dict(p.imap_unordered(get_data, url_list))
    print(results)