我现在正在处理一个大数据集。我的输入将是 4 个不同的数据集，并且我必须对每个数据集应用特定的函数。所以我所做的就是读取所有四个数据集，并使用 pool.map 将函数并行应用到每个数据集。所以现在我有一个父进程和 4 个子进程。直到这一切都很好。
Q1. 现在每个进程内部发生了什么。在我应用于每个数据集的函数中，我将每个元组与其他元组进行比较，因此它是一种递归。有没有办法使其并行，因为这种比较可能需要很长时间，因为数据集很大。如何制作它，因为它已经是一个子进程？是否可以在子进程中再次并行化它，因为我有更多的处理器，所以我想利用它。
Q2。我对于这个递归任务的并行化的想法是，如果我将元组 x 与元组 y 进行比较（每个元组与所有其他元组），我可以为 x 制作块，每个块与 y 进行比较。我想这可以通过两个“for 循环”来完成。有什么建议如何做到这一点？
python parallel-processing multiprocessing python-3.x python-multiprocessing
我正在尝试使用该multiprocessing包来同时读取文件并在进行一些数据转换后覆盖它（部分）。我知道这似乎有点抽象，但我可以使用这种并发性来加速我自己的blocksyncfork。
您可以在下面找到我的代码片段：
#!/usr/bin/python2
import multiprocessing
import sys
import time
blocksize=1024
def do_open(f, mode):
f = open(f, mode)
f.seek(0, 2)
size = f.tell()
f.seek(0)
return f, size
def pipe_getblocks(f, pipe, side):
print "Child file object ID: "+str(id(f))
while True:
print "getblocks_seek_prev: "+str(f.tell())
block = f.read(blocksize)
if not block:
break
print "getblocks_seek_next: "+str(f.tell())
pipe.send(block)
def pipe_server(dev):
f, size = do_open(dev, 'r+')
parent,child = multiprocessing.Pipe(False)
reader = multiprocessing.Process(target=pipe_getblocks, args=(f,child,"R"))
reader.daemon = True
reader.start()
child.close()
i = 0
print …我遇到以下问题。我想实现一个网络爬虫，到目前为止，它可以工作，但速度太慢，所以我尝试使用多重处理来获取 URL。不幸的是我在这个领域不是很有经验。经过一番阅读后，我认为最简单的方法是使用map来自的方法multiprocessing.pool。
但我不断收到以下错误：
TypeError: Pickling an AuthenticationString object is disallowed for security reasons
我发现很少有同样错误的案例，不幸的是它们并没有帮助我。
我创建了代码的精简版本，它可以重现该错误：
import multiprocessing
class TestCrawler:
def __init__(self):
self.m = multiprocessing.Manager()
self.queue = self.m.Queue()
for i in range(50):
self.queue.put(str(i))
self.pool = multiprocessing.Pool(6)
def mainloop(self):
self.process_next_url(self.queue)
while True:
self.pool.map(self.process_next_url, (self.queue,))
def process_next_url(self, queue):
url = queue.get()
print(url)
c = TestCrawler()
c.mainloop()
我将非常感谢任何帮助或建议！
python queue multiprocessing python-3.x python-multiprocessing
以下代码未按预期执行。
import multiprocessing
lock = multiprocessing.Lock()
def dummy():
def log_results_l1(results):
lock.acquire()
print("Writing results", results)
lock.release()
def mp_execute_instance_l1(cmd):
print(cmd)
return cmd
cmds = [x for x in range(10)]
pool = multiprocessing.Pool(processes=8)
for c in cmds:
pool.apply_async(mp_execute_instance_l1, args=(c, ), callback=log_results_l1)
pool.close()
pool.join()
print("done")
dummy()
但是如果函数不是嵌套的，它确实有效。到底是怎么回事。
此代码 ping 各种机器。您能否帮我更改此代码，以便如果 ping 进程挂起超过 7 秒，它会关闭并返回一些标志？
（我想从使用 WMI 的机器中提取各种数据。为此，我将 ping 功能更改为其他功能。问题是在某些机器上 WMI 已损坏并且提取数据的过程无限期挂起。需要超时。）
import multiprocessing.dummy
import subprocess
import numpy as np
import time
start_time = time.time()
def ping(ipadd):
try:
response = subprocess.check_output(['ping', ipadd])
return True
except subprocess.CalledProcessError as e:
return False
#print(ping('10.25.59.20'))
machine_names = \
'''
ya.ru
microsoft.com
www.google.com
www.amazon.com
www.nasa.com
'''.split()
np_machine_names = np.array(machine_names)
p = multiprocessing.dummy.Pool(7)
ping_status = p.map(ping, machine_names)
np_ping_status = np.fromiter(ping_status, dtype=bool)
print(*np_machine_names[np_ping_status], sep = '\n')
run_time = time.time() - start_time
print(f'Runtime: {run_time:.0f}')
更新：虽然我很欣赏关于添加超时到子进程的提示，但问题仍然存在。如何关闭挂起的功能？假设我已将 …
假设我有这个功能：
def f():
while True:
x = generate_something()
if x == condition:
return x
if __name__ == '__main__':
p=Pool(4)
我想在多进程中运行此函数，当其中一个进程满足我的函数条件时，我希望所有其他进程停止。
我有一个循环正在检查计算结果为的值的第一个实例True：
for i in l:
if fun(i):
return i
return None
我想使用多处理池来加快速度，但我的理解是池只会fun在l. 有没有办法在池化时进行某种“短路”？
我正在使用 python 脚本进行大量计算。由于它受 CPU 限制，我通常使用线程模块的方法没有产生任何性能改进。
我现在试图使用多处理而不是多线程来更好地使用我的 CPU 并加速冗长的计算。
我在 stackoverflow 上找到了一些示例代码，但我没有让脚本接受多个参数。有人可以帮我解决这个问题吗？在我很确定我使用 Pool.map 之前我从未使用过这些模块。- 任何帮助表示赞赏。也欢迎使用其他方法来完成多处理。
from multiprocessing import Pool
def calculation(foo, bar, foobar, baz):
# Do a lot of calculations based on the variables
# Later the result is written to a file.
result = foo * bar * foobar * baz
print(result)
if __name__ == '__main__':
for foo in range(3):
for bar in range(5):
for baz in range(4):
for foobar in range(10):
Pool.map(calculation, foo, bar, foobar, baz)
Pool.close()
Pool.join()
我正在编写一个将使用 python 的多处理和线程模块的脚本。为了您的理解，我产生了尽可能多的内核可用的进程，并且在每个进程中我启动了例如 25 个线程。每个线程从一个消费input_queue并产生一个output_queue。对于队列对象，我使用multiprocessing.Queue.
在我的第一次测试之后，我遇到了死锁，因为负责提供和刷新队列的线程挂了。一段时间后，我发现我可以用它Queue().cancel_join_thread()来解决这个问题。
但由于数据丢失的可能性，我想使用： multiprocessing.Manager().Queue()
现在的实际问题是：为每个队列使用一个管理器对象是否更好？或者我应该创建一个管理器并从同一个管理器对象中获取两个问题？
# One manager for all queues
import multiprocessing
manager = multiprocessing.Manager()
input_queue = manager.Queue()
output_queue = manager.Queue()
...Magic...
# As much managers as queues
manager_in = multiprocessing.Manager()
queue_in = manager_in.Queue()
manager_out = multiprocessing.Manager()
queue_out = manager_out.Queue()
...Magic...
感谢您的帮助。
我正在尝试从维基百科文本数据训练 word2vec 模型，为此我使用以下代码。
import logging
import os.path
import sys
import multiprocessing
from gensim.corpora import WikiCorpus
from gensim.models import Word2Vec
from gensim.models.word2vec import LineSentence
if __name__ == '__main__':
program = os.path.basename(sys.argv[0])
logger = logging.getLogger(program)
logging.basicConfig(format='%(asctime)s : %(levelname)s : %(message)s')
logging.root.setLevel(level=logging.INFO)
logger.info("running %s" % ' '.join(sys.argv))
# check and process input arguments
if len(sys.argv) < 3:
print (globals()['__doc__'])
sys.exit(1)
inp, outp = sys.argv[1:3]
model = Word2Vec(LineSentence(inp), size=400, window=5, min_count=5, workers=multiprocessing.cpu_count())
# trim unneeded model memory = use (much) less RAM
model.init_sims(replace=True)
model.save(outp) …python multiprocessing gensim word2vec python-multiprocessing