Mel*_*oun 3 python queue multiprocessing pandas python-multiprocessing
我在通过队列在 2 个进程之间交换对象（数据帧）时遇到了一些麻烦。
第一个进程从队列中获取数据，第二个进程将数据放入队列。put-process 更快，所以 get-process 应该通过读取所有对象来清除队列。
我有奇怪的行为，因为我的代码按预期完美运行，但仅适用于数据帧中的 100 行，对于 1000 行，get-process 始终只需要 1 个对象。
import multiprocessing, time, sys
import pandas as pd
NR_ROWS = 1000
i = 0
def getDf():
global i, NR_ROWS
myheader = ["name", "test2", "test3"]
myrow1 = [ i, i+400, i+250]
df = pd.DataFrame([myrow1]*NR_ROWS, columns = myheader)
i = i+1
return df
def f_put(q):
print "f_put start"
while(1):
data = getDf()
q.put(data)
print "P:", data["name"].iloc[0]
sys.stdout.flush()
time.sleep(1.55)
def f_get(q):
print "f_get start"
while(1):
data = pd.DataFrame()
while not q.empty():
data = q.get()
print "get"
if not data.empty:
print "G:", data["name"].iloc[0]
else:
print "nothing new"
time.sleep(5.9)
if __name__ == "__main__":
q = multiprocessing.Queue()
p = multiprocessing.Process(target=f_put, args=(q,))
p.start()
while(1):
f_get(q)
p.join()
输出 100 行数据帧，get-process 获取所有对象
f_get start
nothing new
f_put start
P: 0 # put 1.object into the queue
P: 1 # put 2.object into the queue
P: 2 # put 3.object into the queue
P: 3 # put 4.object into the queue
get # get-process takes all 4 objects from the queue
get
get
get
G: 3
P: 4
P: 5
P: 6
get
get
get
G: 6
P: 7
P: 8
输出 1000 行数据帧，get-process 只需要一个对象。
f_get start
nothing new
f_put start
P: 0 # put 1.object into the queue
P: 1 # put 2.object into the queue
P: 2 # put 3.object into the queue
P: 3 # put 4.object into the queue
get <-- #!!! get-process takes ONLY 1 object from the queue!!!
G: 1
P: 4
P: 5
P: 6
get
G: 2
P: 7
P: 8
P: 9
P: 10
get
G: 3
P: 11
知道我做错了什么以及如何传递更大的数据框吗？
冒着无法完全提供功能齐全的示例的风险，这里是出了什么问题。
首先，它是一个时间问题。
我用更大的 DataFrames（10000甚至100000）再次尝试了你的代码，我开始看到和你一样的东西。这意味着一旦数组的大小超过某个取决于系统（CPU？）的阈值，您就会看到这种行为。
我稍微修改了您的代码，以便更容易地看到会发生什么。首先，5DataFrames是put为没有任何自定义的队列time.sleep。在f_get函数中，我向time.sleep(0)循环 ( while not q.empty())添加了一个计数器（和 a ，见下文）。
新代码：
import multiprocessing, time, sys
import pandas as pd
NR_ROWS = 10000
i = 0
def getDf():
global i, NR_ROWS
myheader = ["name", "test2", "test3"]
myrow1 = [ i, i+400, i+250]
df = pd.DataFrame([myrow1]*NR_ROWS, columns = myheader)
i = i+1
return df
def f_put(q):
print "f_put start"
j = 0
while(j < 5):
data = getDf()
q.put(data)
print "P:", data["name"].iloc[0]
sys.stdout.flush()
j += 1
def f_get(q):
print "f_get start"
while(1):
data = pd.DataFrame()
loop = 0
while not q.empty():
data = q.get()
print "get (loop: %s)" %loop
time.sleep(0)
loop += 1
time.sleep(1.)
if __name__ == "__main__":
q = multiprocessing.Queue()
p = multiprocessing.Process(target=f_put, args=(q,))
p.start()
while(1):
f_get(q)
p.join()
现在，如果您针对不同的行数运行此程序，您将看到如下内容：
N=100：
f_get start
f_put start
P: 0
P: 1
P: 2
P: 3
P: 4
get (loop: 0)
get (loop: 1)
get (loop: 2)
get (loop: 3)
get (loop: 4)
N=10000：
f_get start
f_put start
P: 0
P: 1
P: 2
P: 3
P: 4
get (loop: 0)
get (loop: 1)
get (loop: 0)
get (loop: 0)
get (loop: 0)
这告诉我们什么？
只要DataFrame很小，您认为该put过程比get看起来更快的假设是正确的，我们就可以在 的一个循环中获取所有 5 个项目while not q.empty()。
但是，随着行数的增加，有些东西会发生变化。while 条件q.empty()评估为True（队列为空）和外部while(1)循环。
这可能意味着put现在比现在慢get，我们必须等待。但是如果我们将整体的睡眠时间设置为f_get类似15，我们仍然会得到相同的行为。
另一方面，如果我们time.sleep(0)将内部q.get()循环中的更改为 1，
while not q.empty():
data = q.get()
time.sleep(1)
print "get (loop: %s)" %loop
loop += 1
我们得到这个：
f_get start
f_put start
P: 0
P: 1
P: 2
P: 3
P: 4
get (loop: 0)
get (loop: 1)
get (loop: 2)
get (loop: 3)
get (loop: 4)
这看起来对！这意味着实际上get做了一些奇怪的事情。似乎在它仍在处理 a 时get，队列状态为empty，并且在get完成后下一项可用。
我确信这是有原因的，但我不够熟悉，multiprocessing无法看到。
根据您的应用程序，您可以将适当的添加time.sleep到您的内部循环中，看看是否足够。
或者，如果您想解决它（而不是使用变通方法作为time.sleep方法），您可以查看multiprocessing并查找有关阻塞、非阻塞或异步通信的信息 - 我认为可以在那里找到解决方案。