标签: multiprocessing

问题

我正在编写一个从语料库中读取一组文档的程序（每一行都是一个文档）。每个文档都使用一个函数进行处理processdocument，分配一个唯一的 ID，然后写入数据库。理想情况下，我们希望使用多个进程来完成此操作。逻辑如下：

1. 主例程创建一个新数据库并设置一些表。
2. 主例程设置一组将运行辅助函数的进程/线程。
3. 主程序启动所有进程。
4. 主程序读取语料库，将文档添加到队列中。
5. 每个进程的工作函数循环，从队列中读取文档，使用 从中提取信息processdocument，并将信息写入数据库表中的新条目。
6. 一旦队列为空并且主例程设置了适当的标志（一旦没有更多文档添加到队列中），工作者循环就会中断。

题

我对 sqlalchemy（以及一般的数据库）比较陌生。据我所知，我认为用于在主例程中设置数据库的代码工作正常。我遇到的问题是，我不确定在不与其他进程发生冲突的情况下，为每个进程写入数据库的工作函数到底应该放什么。

没有什么特别复杂的事情：每个进程都从一个受锁保护的 multiprocessing.Value 对象中获得一个唯一的值来分配给一个条目。我只是不确定是否应该将什么传递给工作函数（队列除外），如果有的话。我是否传递了我在主程序中创建的 sqlalchemy.Engine 实例？元数据实例？我是否为每个流程创建一个新引擎？有没有其他一些规范的方法来做到这一点？有什么特别需要我记住的吗？

补充评论

我很清楚我可以不打扰多处理，而是在一个进程中执行此操作，但是我将不得不编写具有多个进程读取数据库的代码，所以我不妨弄清楚该怎么做现在这个。

在此先感谢您的帮助！

在多处理器架构中，寄存器是如何组织的？

例如，在 4 核处理器中，一次至少可以运行 4 个进程。
堆栈指针、程序状态寄存器和程序计数器是如何组织的？

其他通用寄存器呢？

我的猜测是，每个内核都有一组单独的寄存器。

我想使用多处理模块来加速网页抓取。我的目标是在页面中提取一部分 HTML 并将其保存在父变量中。最后，将该变量写入文件。

但我遇到的问题是处理页面大约需要 1 秒。

我的代码有效，但它没有做我想要的：

import urllib.request
from bs4 import BeautifulSoup
from multiprocessing.dummy import Pool  # This is a thread-based Pool
from multiprocessing import cpu_count


def parseWeb(url):
    page = urllib.request.urlopen(url)
    soup = BeautifulSoup(page)
    h2_tag = soup.find('h2', class_='midashigo')
    return h2_tag

if __name__ == '__main__':
    file = 'links.txt' # each link is on a separate line.
    pool = Pool(cpu_count() * 2)
    with open(file, 'r') as f:
        results = pool.map(parseWeb, f)
    with open('output.txt', 'w', encoding='utf-8') as w:
        w.write(str(results))

如何修改它以赋予它多处理的全部功能？谢谢你。

我有一个multiprocessing.pool.map用于做一些工作的 python 脚本。随着它的进行，它会将内容打印到stdout，对于错误，它会打印到stderr。我决定为每个流都有一个单独的日志文件会很好，经过一些思考后，我应该像这样运行它：

time ./ecisSearch.py 58Ni.conf 4 1 > >(tee stdout.log) 2> >(tee stderr.log >&2)

这为我提供了日志文件并将输出保留在适当的流上。然而问题来了。如果我在没有重定向的情况下运行它，我会得到这个：

$ time ./ecisSearch.py 58Ni.conf 4 1

2015-01-09 14:42:37.524333: This job will perform 4 fit(s)   //this is stdout

2015-01-09 14:42:37.524433: Threaded mapping of the fit function onto the starting point input set is commencing   //this is stdout

2015-01-09 14:42:37.526641: Starting run #: 0   //this is stdout
2015-01-09 14:42:37.527018: Starting run #: 1   //this is stdout
2015-01-09 14:42:37.529124: Starting run …

我是multiprocessingPython的新手(2.7).

我尝试运行以下代码:

from time import sleep
from multiprocessing import Process
import multiprocessing

def func(x):
    print("start %s"%(x))
    sleep(x)
    print("end %s"%(x))
    return

if __name__ == '__main__':

    Process(target=func(10)).start()
    Process(target=func(1)).start()

这返回的东西是:

start 10
end 10
start 1
end 1

虽然start 1并且end 1应该先于先前出现end 10.

我很感激帮助理解我在这里可能缺少的东西.

我很难理解超线程.如果逻辑核心实际上不存在,那么使用超线程有什么意义呢？在维基百科的文章指出:

对于物理上存在的每个处理器核心,操作系统寻址两个虚拟(逻辑)核心,并在可能的情况下共享它们之间的工作负载.

如果两个逻辑内核共享相同的执行单元,这意味着其中一个线程必须被保持而另一个执行,这就是说,我不明白超线程如何有用,因为你是实际上没有引入新的执行单元.我无法绕过这个

我试图在python下运行并行进程(在ubuntu上).

我开始使用多处理,它适用于简单的例子.
然后是泡菜错误,所以我切换到了悲.. 我对不同的选项感到困惑,所以写了一个非常简单的基准测试代码.

import multiprocessing as mp
from pathos.multiprocessing import Pool as Pool1
from pathos.pools import ParallelPool as Pool2
from pathos.parallel import ParallelPool as Pool3
import time

def square(x):  
    # calculate the square of the value of x
    return x*x

if __name__ == '__main__':

    dataset = range(0,10000)

    start_time = time.time()
    for d in dataset:
        square(d)
    print('test with no cores: %s seconds' %(time.time() - start_time))

    nCores = 3
    print('number of cores used: %s' %(nCores))  


    start_time = time.time()

    p = mp.Pool(nCores) …

我有一个视频文件，需要逐帧处理，然后需要在帧中显示结果。目前，我正在按顺序进行处理，并逐帧显示帧。

现在，我想并行而不是顺序地处理帧。一旦处理了X个帧，则必须出现cv2.imshow，并且必须以正确的顺序显示已处理的帧。

目前，我的顺序代码如下所示

import cv2
import requests


def process_frame(bgr_image, jpg_as_text):
    try:
        # Post to api for processing and get the results
        # result = requests.post("example.com", data={"jpg": jpg_as_text})

        # Add results to bgr_image
        # cv2.putText()
    except Exception as e:
        print(e)
        pass
    # Show the frame
    cv2.imshow("frame", bgr_image)


video = cv2.VideoCapture("video.mp4")
i = 0


while video.isOpened():
    ret, bgr_image = video.read()
    if ret == True:
        img_height, img_width, _ = bgr_image.shape
        jpg_as_text = cv2.imencode(".jpg", bgr_image)[1].tostring()
        process_frame(bgr_image, jpg_as_text)
        print(i)
        i += 1
    else:
        break
    if …

这是代码:

一:

import multiprocessing
import datetime
import time
a = [i for i in range(100)]
res1 = []
def jia(a,b,c):
    time.sleep(0.1)
    return a + b + c
res1_time_start = datetime.datetime.now()
p = multiprocessing.Pool(4)
res1 = [p.apply_async(jia,(i,1,1)) for i in a]
res1 = [i.get() for i in res1]
res1_time_end = datetime.datetime.now()
print((res1_time_end - res1_time_start).seconds)

另一个:

import multiprocessing
import datetime
import time
a = [i for i in range(100)]
res1 = []
def jia(a,b,c):
    time.sleep(0.1)
    return a + b + c
res1_time_start …

在文档中，Manager与上下文管理器（即with）一起使用，如下所示：

from multiprocessing.managers import BaseManager

class MathsClass:
    def add(self, x, y):
        return x + y
    def mul(self, x, y):
        return x * y

class MyManager(BaseManager):
    pass

MyManager.register('Maths', MathsClass)

if __name__ == '__main__':
    with MyManager() as manager:
        maths = manager.Maths()
        print(maths.add(4, 3))         # prints 7
        print(maths.mul(7, 8))         # prints 56

但是，除了名称空间之外，这样做还有什么好处？对于打开文件流，其好处是显而易见的，因为您不必手动.close()进行连接，但这对Manager有什么作用？如果您不在上下文中使用它，则必须使用什么步骤来确保所有内容均正确关闭？

简而言之，与上述类似，使用以上内容有什么好处：

manager = MyManager()
maths = manager.Maths()
print(maths.add(4, 3))         # prints 7
print(maths.mul(7, 8))         # prints 56