多处理列表搜索

Question

所以我使用进程和队列来搜索数据并找到在不同列中具有相同条目的行。我决定使用多处理来尝试使其可以针对大数据进行扩展。该文件有 1000 行和每行 10 个数据点。我只读取了 80 行数据并且程序停止了。 70 行，运行良好，速度也不错。

我的问题是我做错了什么，或者我没有发现这种方法的局限性？该代码无论如何都不是完美的，并且本身可能很糟糕。代码如下：

from multiprocessing import Process, Queue
import random

def openFile(file_name, k, division):
    i = 0
    dataSet = []
    with open(file_name) as f:
        for line in f:
            stripLine = line.strip('\n')
            splitLine = stripLine.split(division)
            dataSet += [splitLine]
            i += 1
            if(i == k):
                break

    return(dataSet)

def setCombination(q,data1,data2):
    newData = []
    for i in range(0,len(data1)):
        for j in range(0, len(data2)):
            if(data1[i][1] == data2[j][3]):
                newData += data2[j]
    q.put(newData)

if __name__ == '__main__':
    # Takes in the file, the length of the data to read in, and how the data is divided.
    data = openFile('testing.txt', 80, ' ')
    for i in range(len(data)):
        for j in range(len(data[i])):
            try:
                data[i][j] = float(data[i][j])
            except ValueError:
                 pass

    #print(data)
    k = len(data)//10
    q = Queue()
    processes = [Process(target=setCombination, args=(q, data[k*x: k + k*x], data))
                                                                for x in range(10)]
    for p in processes:
        p.start()

    # Exit the completed processes
    for p in processes:
        p.join()

    saleSet = [q.get() for p in processes]
    print('\n', saleSet)

数据文件testing.txt

Answer 1

您的代码所做的某些事情似乎导致了死锁。在实验过程中，我注意到 10 个任务中有 3 个永远不会终止，但老实说，我真的不知道原因。

好消息是它很容易通过删除或禁用来修复

# Exit the completed processes
for p in processes:
    p.join()

你的代码中有循环。

这是您的代码的完整版本，其中（大部分）只是进行了修改：

from multiprocessing import Process, Queue

def openFile(file_name, k, division):
    i = 0
    dataSet = []
    with open(file_name) as f:
        for line in f:
            stripLine = line.strip('\n')
            splitLine = stripLine.split(division)
            dataSet += [splitLine]
            i += 1
            if i == k:
                break

    return dataSet

def setCombination(q, data1, data2):
    newData = []
    for i in range(len(data1)):
        for j in range(len(data2)):
            if data1[i][1] == data2[j][3]:
                newData += data2[j]
    q.put(newData)

if __name__ == '__main__':
    # Takes in the file, the length of the data to read in, and how the data is divided.
    data = openFile('testing.txt', 80, ' ')

    for i in range(len(data)):
        for j in range(len(data[i])):
            try:
                data[i][j] = float(data[i][j])
            except ValueError:
                 pass

    k = len(data) // 10
    q = Queue()
    processes = [Process(target=setCombination, args=(q, data[k*x: k*x+k], data))
                    for x in range(10)]
    for p in processes:
        p.start()

# NO LONGER USED (HANGS)
#    # Exit the completed processes
#    for p in processes:
#        p.join()

    # note: this works since by default, get() will block until it can retrieve something
    saleSet = [q.get() for _ in processes]  # a queue item should be added by each Process
    print('\n', saleSet)