线程池与多处理池类似吗？
- 2024-12-03 08:45:00
- admin 原创
- 157
问题描述：
是否有一个用于工作线程的 Pool 类，类似于多处理模块的Pool 类？
例如，我喜欢并行化 map 函数的简单方法
def long_running_func(p):
c_func_no_gil(p)
p = multiprocessing.Pool(4)
xs = p.map(long_running_func, range(100))
然而，我想在不产生创建新流程的开销的情况下实现此目的。
我知道 GIL。但是，在我的用例中，该函数将是一个 IO 绑定的 C 函数，python 包装器将在实际函数调用之前释放 GIL。
我必须编写自己的线程池吗？
解决方案 1：
我刚刚发现模块中实际上 有一个基于线程的 Pool 接口multiprocessing，但是它有点隐藏并且没有正确记录。
可以通过以下方式导入
from multiprocessing.pool import ThreadPool
它是使用包装 Python 线程的虚拟 Process 类实现的。可以在文档multiprocessing.dummy中简要提到此基于线程的 Process 类。这个虚拟模块据称提供了基于线程的整个多处理接口。
解决方案 2：
在 Python 3 中你可以使用concurrent.futures.ThreadPoolExecutor，即：
executor = ThreadPoolExecutor(max_workers=10)
a = executor.submit(my_function)
请参阅文档以获取更多信息和示例。
解决方案 3：
是的，并且它似乎具有（或多或少）相同的 API。
import multiprocessing
def worker(lnk):
....
def start_process():
.....
....
if(PROCESS):
pool = multiprocessing.Pool(processes=POOL_SIZE, initializer=start_process)
else:
pool = multiprocessing.pool.ThreadPool(processes=POOL_SIZE,
initializer=start_process)
pool.map(worker, inputs)
....
解决方案 4：
对于一些非常简单和轻量级的东西（从这里稍微修改了一下）：
from Queue import Queue
from threading import Thread
class Worker(Thread):
"""Thread executing tasks from a given tasks queue"""
def __init__(self, tasks):
Thread.__init__(self)
self.tasks = tasks
self.daemon = True
self.start()
def run(self):
while True:
func, args, kargs = self.tasks.get()
try:
func(*args, **kargs)
except Exception, e:
print e
finally:
self.tasks.task_done()
class ThreadPool:
"""Pool of threads consuming tasks from a queue"""
def __init__(self, num_threads):
self.tasks = Queue(num_threads)
for _ in range(num_threads):
Worker(self.tasks)
def add_task(self, func, *args, **kargs):
"""Add a task to the queue"""
self.tasks.put((func, args, kargs))
def wait_completion(self):
"""Wait for completion of all the tasks in the queue"""
self.tasks.join()
if __name__ == '__main__':
from random import randrange
from time import sleep
delays = [randrange(1, 10) for i in range(100)]
def wait_delay(d):
print 'sleeping for (%d)sec' % d
sleep(d)
pool = ThreadPool(20)
for i, d in enumerate(delays):
pool.add_task(wait_delay, d)
pool.wait_completion()
为了支持任务完成时的回调，您只需将回调添加到任务元组即可。
解决方案 5：
您好，要在 Python 中使用线程池，您可以使用这个库：
from multiprocessing.dummy import Pool as ThreadPool
然后为了使用，这个库执行以下操作：
pool = ThreadPool(threads)
results = pool.map(service, tasks)
pool.close()
pool.join()
return results
线程是您想要的线程数，任务是大多数映射到服务的任务列表。
解决方案 6：
是的，有一个类似于多处理池的线程池，但是，它有点隐蔽，没有正确记录。您可以通过以下方式导入它：-
from multiprocessing.pool import ThreadPool
我只是给你展示简单的例子
def test_multithread_stringio_read_csv(self):
# see gh-11786
max_row_range = 10000
num_files = 100
bytes_to_df = [
'
'.join(
['%d,%d,%d' % (i, i, i) for i in range(max_row_range)]
).encode() for j in range(num_files)]
files = [BytesIO(b) for b in bytes_to_df]
# read all files in many threads
pool = ThreadPool(8)
results = pool.map(self.read_csv, files)
first_result = results[0]
for result in results:
tm.assert_frame_equal(first_result, result)
解决方案 7：
这是我最终使用的结果。它是上面 dgorissen 的类的修改版本。
文件：threadpool.py
from queue import Queue, Empty
import threading
from threading import Thread
class Worker(Thread):
_TIMEOUT = 2
""" Thread executing tasks from a given tasks queue. Thread is signalable,
to exit
"""
def __init__(self, tasks, th_num):
Thread.__init__(self)
self.tasks = tasks
self.daemon, self.th_num = True, th_num
self.done = threading.Event()
self.start()
def run(self):
while not self.done.is_set():
try:
func, args, kwargs = self.tasks.get(block=True,
timeout=self._TIMEOUT)
try:
func(*args, **kwargs)
except Exception as e:
print(e)
finally:
self.tasks.task_done()
except Empty as e:
pass
return
def signal_exit(self):
""" Signal to thread to exit """
self.done.set()
class ThreadPool:
"""Pool of threads consuming tasks from a queue"""
def __init__(self, num_threads, tasks=[]):
self.tasks = Queue(num_threads)
self.workers = []
self.done = False
self._init_workers(num_threads)
for task in tasks:
self.tasks.put(task)
def _init_workers(self, num_threads):
for i in range(num_threads):
self.workers.append(Worker(self.tasks, i))
def add_task(self, func, *args, **kwargs):
"""Add a task to the queue"""
self.tasks.put((func, args, kwargs))
def _close_all_threads(self):
""" Signal all threads to exit and lose the references to them """
for workr in self.workers:
workr.signal_exit()
self.workers = []
def wait_completion(self):
"""Wait for completion of all the tasks in the queue"""
self.tasks.join()
def __del__(self):
self._close_all_threads()
def create_task(func, *args, **kwargs):
return (func, args, kwargs)
使用游泳池
from random import randrange
from time import sleep
delays = [randrange(1, 10) for i in range(30)]
def wait_delay(d):
print('sleeping for (%d)sec' % d)
sleep(d)
pool = ThreadPool(20)
for i, d in enumerate(delays):
pool.add_task(wait_delay, d)
pool.wait_completion()
解决方案 8：
另一种方法是将进程添加到线程队列池中
import concurrent.futures
with concurrent.futures.ThreadPoolExecutor(max_workers=cpus) as executor:
for i in range(10):
a = executor.submit(arg1, arg2,....)
解决方案 9：
创建新进程的开销很小，尤其是当只有 4 个进程时。我怀疑这是否是您的应用程序的性能热点。保持简单，优化您需要优化的地方以及分析结果指向的地方。
解决方案 10：
没有内置基于线程的池。但是，使用该类可以非常快速地实现生产者/消费者队列Queue。
来自：
https://docs.python.org/2/library/queue.html
from threading import Thread
from Queue import Queue
def worker():
while True:
item = q.get()
do_work(item)
q.task_done()
q = Queue()
for i in range(num_worker_threads):
t = Thread(target=worker)
t.daemon = True
t.start()
for item in source():
q.put(item)
q.join() # block until all tasks are done
解决方案 11：
如果您不介意执行其他人的代码，这是我的：
注意：您可能需要删除许多额外的代码[添加以便更好地说明和演示其工作原理]
注意：方法名称和变量名称使用 Python 命名约定，而不是驼峰式命名法。
工作流程：
MultiThread 类将通过共享锁、工作队列、退出标志和结果来启动没有线程实例的情况。
一旦创建所有实例，MultiThread 将启动 SingleThread。
我们可以使用多线程添加工作（它将负责锁定）。
SingleThreads 将使用中间的锁来处理工作队列。
一旦你的工作完成，你可以销毁所有具有共享布尔值的线程。
这里的工作可以是任何东西。它可以自动导入（取消注释导入行）并使用给定的参数处理模块。
结果将被添加到结果中，我们可以使用 get_results 获取
代码：
import threading
import queue
class SingleThread(threading.Thread):
def __init__(self, name, work_queue, lock, exit_flag, results):
threading.Thread.__init__(self)
self.name = name
self.work_queue = work_queue
self.lock = lock
self.exit_flag = exit_flag
self.results = results
def run(self):
# print("Coming %s with parameters %s", self.name, self.exit_flag)
while not self.exit_flag:
# print(self.exit_flag)
self.lock.acquire()
if not self.work_queue.empty():
work = self.work_queue.get()
module, operation, args, kwargs = work.module, work.operation, work.args, work.kwargs
self.lock.release()
print("Processing : " + operation + " with parameters " + str(args) + " and " + str(kwargs) + " by " + self.name + "
")
# module = __import__(module_name)
result = str(getattr(module, operation)(*args, **kwargs))
print("Result : " + result + " for operation " + operation + " and input " + str(args) + " " + str(kwargs))
self.results.append(result)
else:
self.lock.release()
# process_work_queue(self.work_queue)
class MultiThread:
def __init__(self, no_of_threads):
self.exit_flag = bool_instance()
self.queue_lock = threading.Lock()
self.threads = []
self.work_queue = queue.Queue()
self.results = []
for index in range(0, no_of_threads):
thread = SingleThread("Thread" + str(index+1), self.work_queue, self.queue_lock, self.exit_flag, self.results)
thread.start()
self.threads.append(thread)
def add_work(self, work):
self.queue_lock.acquire()
self.work_queue._put(work)
self.queue_lock.release()
def destroy(self):
self.exit_flag.value = True
for thread in self.threads:
thread.join()
def get_results(self):
return self.results
class Work:
def __init__(self, module, operation, args, kwargs={}):
self.module = module
self.operation = operation
self.args = args
self.kwargs = kwargs
class SimpleOperations:
def sum(self, *args):
return sum([int(arg) for arg in args])
@staticmethod
def mul(a, b, c=0):
return int(a) * int(b) + int(c)
class bool_instance:
def __init__(self, value=False):
self.value = value
def __setattr__(self, key, value):
if key != "value":
raise AttributeError("Only value can be set!")
if not isinstance(value, bool):
raise AttributeError("Only True/False can be set!")
self.__dict__[key] = value
# super.__setattr__(key, bool(value))
def __bool__(self):
return self.value
if __name__ == "__main__":
multi_thread = MultiThread(5)
multi_thread.add_work(Work(SimpleOperations(), "mul", [2, 3], {"c":4}))
while True:
data_input = input()
if data_input == "":
pass
elif data_input == "break":
break
else:
work = data_input.split()
multi_thread.add_work(Work(SimpleOperations(), work[0], work[1:], {}))
multi_thread.destroy()
print(multi_thread.get_results())
相关推荐
热门文章
项目管理软件有哪些？
热门标签
云禅道AD
