主页 / 问题 / 1502522
Accepted
iKey
Asked:2023-03-06 14:44:04 +0000 UTC2023-03-06 14:44:04 +0000 UTC

如何将线程中执行函数的结果写入数组？

  • 772

我找到了一段演示 python 3 中多线程工作的代码。我想让我的函数在每个线程中执行，在所有线程完成后，我得到一个包含所有线程结果的数组

import queue
import threading
import time


# The queue for tasks
q = queue.Queue()

# Worker, handles each task
def worker():
    while True:
        item = q.get()
        if item is None:
            break
        
        #здесь должна выполниться моя функция я записать результат в массив
        result = subprocess.run(['command', 'arg1', 'arg2'], stdout=subprocess.PIPE)

        q.task_done()


def start_workers(worker_pool=1000):
    threads = []
    for i in range(worker_pool):
        t = threading.Thread(target=worker)
        t.start()
        threads.append(t)
    return threads


def stop_workers(threads):
    # stop workers
    for i in threads:
        q.put(None)
    for t in threads:
        t.join()


def create_queue(task_items):
    for item in task_items:
        q.put(item)


if __name__ == "__main__":
    # Dummy tasks
    tasks = [item for item in range(10)]

    # Start up your workers
    workers = start_workers(worker_pool=10)
    create_queue(tasks)

    # Blocks until all tasks are complete
    result = q.join()
    print(result)

    result2 = worker.join()
    print(result2)

    stop_workers(workers)
python

3 个回答

  1. multiprocessing.dummy- 线程接口，重复 api multiprocessing。该函数map大大简化了数组的处理。

    from multiprocessing.dummy import Pool

pool = Pool(1000)

def runner(item):
    ...
    result = ...
    return result

if __name__ == "__main__":
    results = pool.map(runner, range(10))

    队列、等待、运行的 worker 已经实现。

    • 3
  2. Best Answer

    最简单的方法可能是创建另一个队列来存储结果（队列是线程安全的）：

    q = queue.Queue()
# Для хранения результатов
result = queue.Queue()

def worker():
        ...
        # Помещаем результат в очередь
        result.put(subprocess.run(['command', 'arg1', 'arg2'], stdout=subprocess.PIPE))
...
stop_workers(workers)

# Разбираем результаты
for item in result.queue:
    print(item)
    • 2

  3. 通常，不等到 pool 中的所有任务都完成pool.map，然后才整体处理所有结果，而是在每个任务完成后立即开始单独处理结果，与池中正在进行的工作。pool.imap_unordered将不按照我们添加任务的顺序返回结果，哪个任务完成得更快，第一个的结果将被返回。因此，您可以附上进度条，或加快结果的处理速度。

    例如下面，线程池中将向 Google 发送 10 个请求，对请求的响应将保存到文件中。脚本的输出显示保存文件（处理池操作的结果）与查询池的操作并行发生。

    import multiprocessing.pool, time, random, urllib.request

Headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 6.1; WOW64; rv:23.0) Gecko/20100101 Firefox/23.0'}

def runner(word):
    print(f"\n-> {time.strftime('%H:%M:%S', time.gmtime())} search {word}")
    time.sleep(random.randint(1, 5))
    with urllib.request.urlopen(urllib.request.Request(f'https://www.google.com/search?q={word}', headers=Headers)) as http_body:
        return word, http_body.read()

if __name__ == '__main__':
    [*search_words] = range(10)
    with multiprocessing.pool.ThreadPool(processes=4) as pool:
        for word, http_body in pool.imap_unordered(runner, search_words):
            print(f"\n<- {time.strftime('%H:%M:%S', time.gmtime())} save {word}, len={len(http_body)}")
            with open(f'{word}_http_body.html', 'wb') as f:
                f.write(http_body)

    出去：

    -> 20:55:38 search 0
-> 20:55:38 search 1
-> 20:55:38 search 2
-> 20:55:38 search 3
-> 20:55:41 search 4
<- 20:55:41 save 0, len=82176
-> 20:55:42 search 5
<- 20:55:42 save 2, len=78236
-> 20:55:43 search 6
<- 20:55:43 save 1, len=117082
-> 20:55:43 search 7
<- 20:55:43 save 3, len=147944
-> 20:55:44 search 8
<- 20:55:44 save 4, len=83972
-> 20:55:45 search 9
<- 20:55:45 save 5, len=72776
<- 20:55:46 save 8, len=138741
<- 20:55:48 save 6, len=83061
<- 20:55:49 save 7, len=114936
<- 20:55:49 save 9, len=191621
    • 0

相关问题