下面的代码似乎没有并发运行,我不确定具体原因:
def run_normalizers(config, debug, num_threads, name=None):
def _run():
print('Started process for normalizer')
sqla_engine = init_sqla_from_config(config)
image_vfs = create_s3vfs_from_config(config, config.AWS_S3_IMAGE_BUCKET)
storage_vfs = create_s3vfs_from_config(config, config.AWS_S3_STORAGE_BUCKET)
pp = PipedPiper(config, image_vfs, storage_vfs, debug=debug)
if name:
pp.run_pipeline_normalizers(name)
else:
pp.run_all_normalizers()
print('Normalizer process complete')
threads = []
for i in range(num_threads):
threads.append(multiprocessing.Process(target=_run))
[t.start() for t in threads]
[t.join() for t in threads]
run_normalizers(...)
config
变量只是在 _run()
函数之外定义的字典。所有进程似乎都已创建 - 但它并不比我使用单个进程创建速度更快。基本上 run_**_normalizers()
函数中发生的事情是从数据库 (SQLAlchemy) 中的队列表中读取数据,然后发出一些 HTTP 请求,然后运行规范化程序的“管道”来修改数据,然后将其存回数据库。我来自线程“繁重”且经常用于并行处理的 JVM 领域 - 我对此感到有点困惑,因为我认为多进程模块应该绕过 Python 的 GIL 的限制。
最佳答案
修复了我的多处理问题 - 并实际切换了线程。不确定实际上是什么解决了它的想法 - 我只是重新设计了所有东西并制作了 worker 和任务以及没有的东西,现在一切都在飞翔。这是我所做的基础知识:
import abc
from Queue import Empty, Queue
from threading import Thread
class AbstractTask(object):
"""
The base task
"""
__metaclass__ = abc.ABCMeta
@abc.abstractmethod
def run_task(self):
pass
class TaskRunner(object):
def __init__(self, queue_size, num_threads=1, stop_on_exception=False):
super(TaskRunner, self).__init__()
self.queue = Queue(queue_size)
self.execute_tasks = True
self.stop_on_exception = stop_on_exception
# create a worker
def _worker():
while self.execute_tasks:
# get a task
task = None
try:
task = self.queue.get(False, 1)
except Empty:
continue
# execute the task
failed = True
try:
task.run_task()
failed = False
finally:
if failed and self.stop_on_exception:
print('Stopping due to exception')
self.execute_tasks = False
self.queue.task_done()
# start threads
for i in range(0, int(num_threads)):
t = Thread(target=_worker)
t.daemon = True
t.start()
def add_task(self, task, block=True, timeout=None):
"""
Adds a task
"""
if not self.execute_tasks:
raise Exception('TaskRunner is not accepting tasks')
self.queue.put(task, block, timeout)
def wait_for_tasks(self):
"""
Waits for tasks to complete
"""
if not self.execute_tasks:
raise Exception('TaskRunner is not accepting tasks')
self.queue.join()
我所做的就是创建一个 TaskRunner 并向其中添加任务(数千个),然后调用 wait_for_tasks()。所以,很明显,在我所做的重新架构中,我“修复”了我遇到的其他一些问题。不过很奇怪。
关于python - 无法让多处理同时运行进程,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/17738141/