线程池类似于多处理池?

没有内置的基于线程的池。然而，使用Queue类实现生产者/消费者队列可以非常快。

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

创建新进程的开销是最小的，特别是当只有4个进程时。我怀疑这是您的应用程序的性能热点。保持简单，优化你必须做的和分析结果指向的地方。

我刚刚发现在multiprocessing模块中实际上有是一个基于线程的池接口，但是它有些隐藏，没有正确地记录。

它可以通过

from multiprocessing.pool import ThreadPool

它是使用封装python线程的虚拟Process类实现的。这个基于线程的Process类可以在multiprocessing.dummy中找到，它在文档中简要地提到过。这个虚拟模块提供了基于线程的整个多处理接口。

对于一些非常简单和轻量级的东西(从在这里略微修改):

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()

要在任务完成时支持回调，只需将回调添加到任务元组。

在Python 3中，你可以使用concurrent.futures.ThreadPoolExecutor，即:

executor = ThreadPoolExecutor(max_workers=10)
a = executor.submit(my_function)

更多信息和示例请参见文档。

是的，它似乎(或多或少)具有相同的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)
....

嗨，在Python中使用线程池，你可以使用这个库:

from multiprocessing.dummy import Pool as ThreadPool

为了使用，这个库是这样的:

pool = ThreadPool(threads)
results = pool.map(service, tasks)
pool.close()
pool.join()
return results

线程是您想要的线程数，任务是大多数映射到服务的任务列表。

这是我最终使用的结果。它是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()

另一种方法是将进程添加到线程队列池中

import concurrent.futures
with concurrent.futures.ThreadPoolExecutor(max_workers=cpus) as executor:
for i in range(10):
a = executor.submit(arg1, arg2,....)

是的，有一个线程池类似于多处理池，但是，它有些隐藏，没有适当的文档。您可以通过以下方式导入:-

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 = [
'\n'.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)

如果你不介意执行别人的代码，这里是我的:

注意:有很多额外的代码你可能想要删除[添加是为了更好地说明和演示它是如何工作的]

注意: Python命名约定用于方法名和变量名，而不是camelCase。

工作过程:

1. 通过共享锁、工作队列、退出标志和结果，多线程类将启动没有线程实例。
2. 一旦多线程创建了所有实例，SingleThread将被启动。
3. 我们可以使用多线程添加作品(它会处理锁定)。
4. SingleThreads将使用中间的锁来处理工作队列。
5. 工作完成后，可以销毁所有具有共享布尔值的线程。
6. 在这里，工作可以是任何东西。它可以使用给定的参数自动导入(uncomment 进口 line)和处理模块。
7. 结果将被添加到结果中，我们可以使用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 + "\n")
# 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())