建立在 script from this answer 上，我有以下场景:一个包含 2500 个大文本文件(每个约 55Mb)的文件夹，所有文件均以制表符分隔。基本上是 Web 日志。

我需要对每个文件的每一行中的第二个“列”进行 md5 散列，将修改后的文件保存在别处。源文件在机械磁盘上​​，目标文件在 SSD 上。

脚本处理前 25 个(左右)文件的速度非常快。然后它会慢下来。根据前 25 个文件，它应该在 2 分钟左右的时间内完成所有这些文件。但是，根据之后的表现，全部完成需要 15 分钟左右。

它在具有 32 Gb RAM 的服务器上运行，任务管理器很少显示超过 6 Gb 的使用情况。我将其设置为启动 6 个进程，但内核的 CPU 使用率很低，很少超过 15%。

为什么会变慢？磁盘的读/写问题？垃圾收集器？糟糕的代码？关于如何加快速度的任何想法？

这是脚本

import os

import multiprocessing
from multiprocessing import Process
import threading
import hashlib

class ThreadRunner(threading.Thread):
    """ This class represents a single instance of a running thread"""
    def __init__(self, fileset, filedirectory):
        threading.Thread.__init__(self)
        self.files_to_process = fileset
        self.filedir          = filedirectory

    def run(self):
        for current_file in self.files_to_process:

            # Open the current file as read only
            active_file_name = self.filedir + "/" + current_file
            output_file_name = "D:/hashed_data/" + "hashed_" + current_file

            active_file = open(active_file_name, "r")
            output_file = open(output_file_name, "ab+")

            for line in active_file:
                # Load the line, hash the username, save the line
                lineList = line.split("\t")

                if not lineList[1] == "-":
                    lineList[1] = hashlib.md5(lineList[1]).hexdigest()

                lineOut = '\t'.join(lineList)
                output_file.write(lineOut)

            # Always close files after you open them
            active_file.close()
            output_file.close()

            print "\nCompleted " + current_file

class ProcessRunner:
    """ This class represents a single instance of a running process """
    def runp(self, pid, numThreads, fileset, filedirectory):
        mythreads = []
        for tid in range(numThreads):
            th = ThreadRunner(fileset, filedirectory)
            mythreads.append(th) 
        for i in mythreads:
            i.start()
        for i in mythreads:
            i.join()

class ParallelExtractor:    
    def runInParallel(self, numProcesses, numThreads, filedirectory):
        myprocs = []
        prunner = ProcessRunner()

        # Store the file names from that directory in a list that we can iterate
        file_names = os.listdir(filedirectory)

        file_sets = []
        for i in range(numProcesses):
            file_sets.append([])

        for index, name in enumerate(file_names):
            num = index % numProcesses
            file_sets[num].append(name)


        for pid in range(numProcesses):
            pr = Process(target=prunner.runp, args=(pid, numThreads, file_sets[pid], filedirectory)) 
            myprocs.append(pr) 
        for i in myprocs:
            i.start()

        for i in myprocs:
            i.join()

if __name__ == '__main__':    

    file_directory = "E:/original_data"

    processes = 6
    threads   = 1

    extractor = ParallelExtractor()
    extractor.runInParallel(numProcesses=processes, numThreads=threads, filedirectory=file_directory)

最佳答案

散列是一项相对简单的任务，与旋转磁盘的速度相比，现代 CPU 的速度非常快。 i7 上的快速基准测试表明它可以使用 MD5 散列大约 450 MB/s，或者使用 SHA-1 散列 290 MB/s。相比之下，旋转磁盘的典型(顺序原始读取)速度约为 70-150 MB/s。这意味着，即使忽略文件系统的开销和最终的磁盘寻道，CPU 散列文件的速度也比磁盘读取文件的速度快大约 3 倍。

您在处理第一个文件时获得的性能提升可能是因为第一个文件被操作系统缓存在内存中，所以没有磁盘 I/O 发生。这可以通过以下任一方式确认:

• 重启服务器，从而刷新缓存
• 通过从磁盘读取足够大的文件，用其他东西填充缓存
• 在处理第一个文件时仔细聆听是否存在磁盘寻道

现在，由于哈希文件的性能瓶颈是磁盘，因此在多个进程或线程中执行哈希是没有用的，因为它们都将使用同一个磁盘。正如@Max Noel 所提到的，它实际上可以降低 性能，因为您将并行读取多个文件，因此您的磁盘将不得不在文件之间进行查找。正如他所提到的，性能也会因您使用的操作系统的 I/O 调度程序而异。

现在，如果您仍在生成数据，您有一些可能的解决方案:

• 按照@Max Noel 的建议，使用更快的磁盘或 SSD。
• 从多个磁盘中读取 - 在不同的文件系统中或在 RAID 上的单个文件系统中
• 将任务拆分到多台机器上(每台机器有一个或多个磁盘)

但是，如果您只想对这 2500 个文件进行哈希处理，并且您已经将它们放在一个磁盘上，那么这些解决方案就毫无用处。将它们从磁盘读取到其他磁盘然后执行散列较慢，因为您将读取文件两次，并且您可以尽可能快地散列他们。

最后，根据@yaccz 的想法，我想如果您安装了 find、xargs< 的="" cygwin=""> 和 md5sum。

关于python - 为什么这个 python 多处理脚本会在一段时间后变慢？，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/20383114/