python – 处理来自多个进程的单个文件

内容导读

互联网集市收集整理的这篇技术教程文章主要介绍了python – 处理来自多个进程的单个文件，小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含2819字，纯文字阅读大概需要5分钟。

内容图文

我有一个大文本文件,我想处理每一行(做一些操作)并将它们存储在数据库中.由于单个简单程序花费的时间太长,我希望它可以通过多个进程或线程来完成.
每个线程/进程应从该单个文件中读取不同的数据(不同的行),并对其数据(行)执行一些操作并将它们放入数据库中,以便最终处理完所有数据并进行处理.数据库与我需要的数据一起转储.

但我无法弄清楚如何处理这个问题.

解决方法:

您正在寻找的是生产者/消费者模式

基本线程示例

这是使用threading module的基本示例(而不是多处理)

import threading
import Queue
import sys

def do_work(in_queue, out_queue):
    while True:
        item = in_queue.get()
        # process
        result = item
        out_queue.put(result)
        in_queue.task_done()

if __name__ == "__main__":
    work = Queue.Queue()
    results = Queue.Queue()
    total = 20

    # start for workers
    for i in xrange(4):
        t = threading.Thread(target=do_work, args=(work, results))
        t.daemon = True
        t.start()

    # produce data
    for i in xrange(total):
        work.put(i)

    work.join()

    # get the results
    for i in xrange(total):
        print results.get()

    sys.exit()

您不会与线程共享文件对象.您可以通过向queue提供数据行来为他们开展工作.然后每个线程将获取一行,处理它,然后将其返回队列中.

multiprocessing module中内置了一些更高级的工具来共享数据,例如列表和special kind of Queue.使用多处理与线程之间需要权衡取决于你的工作是cpu绑定还是IO绑定.

基本的多处理.Pool示例

这是一个多处理池的基本示例

from multiprocessing import Pool

def process_line(line):
    return "FOO: %s" % line

if __name__ == "__main__":
    pool = Pool(4)
    with open('file.txt') as source_file:
        # chunk the work into batches of 4 lines at a time
        results = pool.map(process_line, source_file, 4)

    print results

A Pool是管理其自身进程的便利对象.由于打开的文件可以遍历其行,因此您可以将其传递给pool.map(),它将循环遍历它并将行传递给worker函数. Map阻塞并在完成后返回整个结果.请注意,这是一个过于简化的示例,并且pool.map()将在完成工作之前将您的整个文件一次性读入内存.如果您希望有大文件,请记住这一点.有更先进的方法来设计生产者/消费者设置.

手动“池”,带限制和行重新排序

这是Pool.map的一个手动示例,但您可以设置一个队列大小,而不是一次性消耗整个迭代,而只是按照它可以处理的速度一个接一个地提供它.我还添加了行号,以便您可以跟踪它们并在以后根据需要引用它们.

from multiprocessing import Process, Manager
import time
import itertools 

def do_work(in_queue, out_list):
    while True:
        item = in_queue.get()
        line_no, line = item

        # exit signal 
        if line == None:
            return

        # fake work
        time.sleep(.5)
        result = (line_no, line)

        out_list.append(result)


if __name__ == "__main__":
    num_workers = 4

    manager = Manager()
    results = manager.list()
    work = manager.Queue(num_workers)

    # start for workers    
    pool = []
    for i in xrange(num_workers):
        p = Process(target=do_work, args=(work, results))
        p.start()
        pool.append(p)

    # produce data
    with open("source.txt") as f:
        iters = itertools.chain(f, (None,)*num_workers)
        for num_and_line in enumerate(iters):
            work.put(num_and_line)

    for p in pool:
        p.join()

    # get the results
    # example:  [(1, "foo"), (10, "bar"), (0, "start")]
    print sorted(results)

内容总结

以上是互联网集市为您收集整理的python – 处理来自多个进程的单个文件全部内容，希望文章能够帮你解决python – 处理来自多个进程的单个文件所遇到的程序开发问题。 如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。

内容备注

版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。

内容手机端