python - 如何在 python 中异步处理 xml？

Question

我有一个很大的 XML 数据文件 (>160M) 需要处理，SAX/expat/pulldom 解析似乎是可行的方法。我想要一个线程筛选节点并将要处理的节点推送到队列中，然后其他工作线程从队列中拉出下一个可用节点并进行处理。

我有以下内容(它应该有锁，我知道 - 稍后会有)

import sys, time
import xml.parsers.expat
import threading

q = []

def start_handler(name, attrs):
    q.append(name)

def do_expat():
    p = xml.parsers.expat.ParserCreate()
    p.StartElementHandler = start_handler
    p.buffer_text = True
    print("opening {0}".format(sys.argv[1]))
    with open(sys.argv[1]) as f:
        print("file is open")
        p.ParseFile(f)
        print("parsing complete")


t = threading.Thread(group=None, target=do_expat)
t.start()

while True:
    print(q)
    time.sleep(1)

问题是 while block 的主体只被调用一次，然后我什至不能 ctrl-C 中断它。在较小的文件上，输出符合预期，但这似乎表明处理程序仅在文档被完全解析时被调用，这似乎违背了 SAX 解析器的目的。

我敢肯定这是我自己的无知，但我不明白我在哪里犯了错误。

PS:我也试过这样改变start_handler:

def start_handler(name, attrs):
    def app():
        q.append(name)
    u = threading.Thread(group=None, target=app)
    u.start()

虽然没有爱。

Answer 1

ParseFile ，正如您所注意到的，只是“吞下”所有东西——对您想要进行的增量解析没有好处!因此，只需一次将文件一点一点地提供给解析器，确保在执行过程中有条件地将控制权交给其他线程——例如:

while True:
  data = f.read(BUFSIZE)
  if not data:
    p.Parse('', True)
    break
  p.Parse(data, False)
  time.sleep(0.0)

time.sleep(0.0) 调用是 Python 的一种方式，它表示“如果有线程准备就绪并正在等待，则让步给其他线程”； Parse 方法已记录 here .

第二点是，忘记这个用法的锁! -- 使用 Queue.Queue相反，它本质上是线程安全的，并且几乎总是在 Python 中协调多线程的最佳和最简单的方法。只需在其上创建一个Queue实例q，q.put(name)，并在q.get( ) 等待更多工作要做 -- 非常简单!

(有几种辅助策略可以用来协调工作线程在没有更多工作要做时的终止，但最简单、没有特殊要求的是只让它们成为守护线程，这样它们就会全部终止当主线程执行时——请参阅 the docs)。