对源自循环的多个线程进行排队的最安全方法是什么？

如何解决对源自循环的多个线程进行排队的最安全方法是什么？

我的脚本遍历输入文件的每一行，并使用每一行中的字符串执行一些操作。由于在每一行执行的任务是相互独立的，因此我决定将任务分成threads，以便脚本不必等待任务完成就可以继续循环。代码如下。


def myFunction(line,param):
    # Doing something with line and param
    # Sends multiple HTTP requests and parse the response and produce outputs
    # Returns nothing

param = arg[1]   
with open(targets,"r") as listfile:
    for line in listfile:
        print("Starting a thread for: ",line)
        t=threading.Thread(target=myFunction,args=(line,param,)) 
        threads.append(t)
        t.start()

我意识到这是一个坏主意，因为输入文件中的行数越来越大。使用此代码，线程数将与行数一样多。经过研究后发现queues就是这样。

我想了解在这种情况下以及是否有其他可以使用的替代方法使用队列的最佳方法。

解决方法

要解决此问题，您可以使用“线程池”的概念，在其中定义要使用的固定数量的线程/工人，例如5个工人，每当一个线程完成执行时，另一个Future（ly）线程将自动替换。

示例：

<Window.DataContext>
    <viewmodel:ViewModelBase/>
</Window.DataContext>

队列是做到这一点的一种方法。使用它们的方法是将函数参数放在队列中，然后使用线程来获取它们并进行处理。

在这种情况下，队列大小并不重要，因为读取下一行很快。在另一种情况下，更优化的解决方案是将队列大小设置为线程数的至少两倍。这样，如果所有线程同时完成对队列中项目的处理，则它们都将准备处理队列中的下一个项目。

为避免复杂化，可以将代码线程设置为守护程序，以便在处理完成后它们不会停止程序的完成。当主进程完成时，它们将被终止。

另一种方法是为每个线程在队列上放置一个特殊项目（例如None），并在从队列中获取线程后使线程退出，然后加入线程。

在下面的示例中，使用workers变量设置工作线程的数量。

这是使用队列的解决方案的示例。

from queue import Queue
from threading import Thread

queue = Queue(workers * 2)
def work():
    while True:
        myFunction(*queue.get())
        queue.task_done()

for _ in range(workers):
    Thread(target=work,daemon=True).start()

with open(targets,'r') as listfile:
    for line in listfile:
        queue.put((line,param))
queue.join()

一个更简单的解决方案可能是使用ThreadPoolExecutor。在这种情况下，这特别简单，因为被调用的函数不会返回需要在主线程中使用的任何内容。

from concurrent.futures import ThreadPoolExecutor

with ThreadPoolExecutor(max_workers=workers) as executor:
    with open(targets,'r') as listfile:
        for line in listfile:
            executor.submit(myFunction,line,param)

此外，如果将所有行都存储在内存中不是问题，那么有一种解决方案，除了线程之外，不使用其他任何东西。工作以这样的方式拆分：线程从列表中读取某些行，而忽略其他行。一个有两个线程的简单示例，其中一个线程读取奇数行，另一个线程读取偶数行。

from threading import Thread

with open(targets,'r') as listfile:
    lines = listfile.readlines()

def work_split(n):
    for line in lines[n::workers]:
        myFunction(line,param)

threads = []
for n in range(workers):
    t = Thread(target=work_split,args=(n,))
    t.start()
    threads.append(t)

for t in threads:
    t.join()

我已经做了一个快速基准测试，Queue比ThreadPoolExecutor快一点，但是拆分工作的解决方案比两者都快。

根据您报告的代码，没有意义使用线程。这是因为没有任何I / O操作，因此线程以线性方式执行而没有多线程。在这种情况下，线程永远不会释放GIL（全局解释器锁定），因此应用程序显然只是在使用多线程，实际上，解释器在程序中仅使用一个CPU，在同一时间使用一个线程。这样，您在使用线程方面没有任何优势，相反，由于切换上下文以及线程启动时线程初始化的开销，这种情况下性能可能会降低。

在这种情况下（如果适用的话），具有更好性能的唯一方法是多进程程序。但是请注意您启动的进程数，请记住每个进程都有自己的解释器。

这是GitFront的一个很好的答案。这个答案只是使用多处理程序包增加了一个选项。使用并发或未来处理取决于特定要求。相对而言，多处理有更多选择，但是对于给定的问题，在最简单的情况下结果应该几乎相同。

from multiprocessing import cpu_count,Pool
PROCESSES = cpu_count() # Warning: uses all cores

def pool_method(listfile,param):
    p = Pool(processes=PROCESSES)
    checker = [p.apply_async(myFunction,(line,param)) for line in listfile]

...

除了“ apply_async”之外，还有其他各种方法，但这应该可以很好地满足您的需求。

对源自循环的多个线程进行排队的最安全方法是什么？

如何解决对源自循环的多个线程进行排队的最安全方法是什么？

解决方法

相关推荐