标签: python-multiprocessing

在 Python 2.7 中处理 CSV 文件时，我在使用ThreadPools 和 aGenerator时遇到问题。下面是一些示例代码，说明了我的观点：

from multiprocessing.dummy import Pool as ThreadPool
import time

def getNextBatch():
    # Reads lines from a huge CSV and yields them as required.
    for i in range(5):
        yield i;

def processBatch(batch):
    # This simulates a slow network request that happens.
    time.sleep(1);
    print "Processed Batch " + str(batch);

# We use 4 threads to attempt to aleviate the bottleneck caused by network I/O.
threadPool = ThreadPool(processes = 4)

batchGenerator = …

from threading import Thread
import time
class ThreadTest():

    def loop1(self):
        for i in range(0, 100, 5):
            print(i)
            time.sleep(2)

    def loop2(self):
        for i in range(100, 210, 11):
            print(i)
            time.sleep(2)

if __name__ == '__main__':
    T1 = Thread(target=ThreadTest().loop1(), args=())
    T2 = Thread(target=ThreadTest().loop2(), args=())
    T1.start()
    T2.start()
    T1.join()
    T2.join()

上面的代码按顺序而不是同时运行这些方法。我想要一种可以同时运行两种方法/进程的方法。

此代码在常规 CPython 3.5 下运行良好：

import concurrent.futures

def job(text):
    print(text)

with concurrent.futures.ProcessPoolExecutor(1) as pool:
    pool.submit(job, "hello")

但是如果你把它作为 运行python -m doctest myfile.py，它就会挂起。更改submit(job为submit(print使其不会挂起，使用ThreadPoolExecutor代替ProcessPoolExecutor.

为什么在 doctest 下运行时会挂起？

我一直在看以下问题，但没有任何运气：

Python 在并行进程之间共享字典

多处理：在进程之间共享一个大的只读对象？

python中的多处理-在多个进程之间共享大对象（例如pandas数据帧）

我编写了一个非常基本的测试文件来说明我正在尝试做的事情：

from collections import deque
from multiprocessing import Process
import numpy as np


class TestClass:
    def __init__(self):
        self.mem = deque(maxlen=4)
        self.process = Process(target=self.run)

    def run(self):
        while True:
            self.mem.append(np.array([0, 1, 2, 3, 4]))


def print_values(x):
    while True:
        print(x)


test = TestClass()
process = Process(target=print_values(test.mem))

test.process.start()
process.start()

目前这输出以下内容：

deque([], maxlen=4)

如何从主代码或运行“print_values”的进程访问内存值？

在python 2中，multiprocessing.dummy.Pool和之间有什么区别multiprocessing.pool.ThreadPool吗？源代码似乎暗示它们是相同的。

我想将共享 DataFrame 的只读访问权限授予由multiprocessing.Pool.map().

我想避免复制和酸洗。

我知道可以使用 pyarrow 。但是，我发现他们的文档非常繁琐。任何人都可以提供一个关于如何完成的例子吗？

假设我有下面的代码，一个执行某些操作的函数，它在进程中启动，并返回一个值。

from multiprocessing import Process

def my_func(arg):
    return 'Hello, ' + arg

p1 = Process(target=my_func, args=('John',)
p1.start()
p1.join()

如何获取函数的返回值？

为什么下面的代码只适用于multiprocessing.dummy，而不适用于 simple multiprocessing。

import urllib.request
#from multiprocessing.dummy import Pool #this works
from multiprocessing import Pool

urls = ['http://www.python.org', 'http://www.yahoo.com','http://www.scala.org', 'http://www.google.com']

if __name__ == '__main__':
    with Pool(5) as p:
        results = p.map(urllib.request.urlopen, urls)

错误 ：

Traceback (most recent call last):
  File "urlthreads.py", line 31, in <module>
    results = p.map(urllib.request.urlopen, urls)
  File "C:\Users\patri\Anaconda3\lib\multiprocessing\pool.py", line 268, in map
    return self._map_async(func, iterable, mapstar, chunksize).get()
  File "C:\Users\patri\Anaconda3\lib\multiprocessing\pool.py", line 657, in get
    raise self._value
multiprocessing.pool.MaybeEncodingError: Error sending result: '[<http.client.HTTPResponse object at 0x0000016AEF204198>]'. Reason: …

这是我想要并行运行 printRange() 的简单代码：

def printRange(lrange):
    print ("First is " + str(lrange[0]) + " and last is " + str(lrange[1]))


def runInParallel():
    ranges = [[0, 10], [10, 20], [20, 30]]
    // Call printRange in parallel with each sublist of ranges given as argument

我的问题与这里的 SO 问题不同，每个过程都是硬编码的，开始并最终加入。我想与其他 100 个 printRange() 工作函数并行运行 printRange()。每次硬编码是不可行的。这怎么可能？

我知道使用传统的多处理库我可以声明一个值并在进程之间共享状态。

https://docs.python.org/3/library/multiprocessing.html?highlight=multiprocessing#sharing-state-between-processes

使用较新的concurrent.futures库时，如何在进程之间共享状态？

import concurrent.futures

def get_user_object(batch):
    # do some work
    counter = counter + 1
    print(counter)

def do_multithreading(batches):
    with concurrent.futures.ThreadPoolExecutor(max_workers=25) as executor:
        threadingResult = executor.map(get_user_object, batches)

def run():
    data_pools = get_data()
    start = time.time()
    with concurrent.futures.ProcessPoolExecutor(max_workers=PROCESSES) as executor:
        processResult = executor.map(do_multithreading, data_pools)
    end = time.time()
    print("TIME TAKEN:", end - start)

if __name__ == '__main__':
    run()

我想保持这个计数器的同步值。

在以前的库中，我可能使用过multiprocessing.Value和Lock.