python - Python线程传递状态

Question

基本上，我想做的是使用代理获取几个网站并处理数据。问题在于请求很少以令人信服的方式失败，设置套接字超时并没有太大帮助，因为它们经常无法正常工作。

所以我所做的是:

q = Queue()
s = ['google.com','ebay.com',] # And so on
for item in s:
    q.put(item)


def worker():
        item = q.get()
        data = fetch(item) # This is the buggy part
        # Process the data, yadayada

for i in range(workers):
    t = InterruptableThread(target=worker)
    t.start()


# Somewhere else
if WorkerHasLivedLongerThanTimeout:
    worker.terminate()

(InterruptableThread class)
问题是我只想杀死仍停留在获取部分上的线程。另外，我希望项目返回队列。 IE:

def worker():
        self.status = 0
        item = q.get()
        data = fetch(item) # This is the buggy part
        self.status = 1 # Don't kill me now, bro!
        # Process the data, yadayada

# Somewhere else
if WorkerHasLivedLongerThanTimeout and worker.status != 1:
    q.put(worker.item)
    worker.terminate()

如何才能做到这一点？

Answer 1

编辑:突发新闻；见下文 · · · ······

我最近决定，我想做类似的事情，结果是pqueue_fetcher模块。最终它主要是一种学习上的努力:我了解到，除其他外，使用twisted之类的东西几乎肯定比尝试以任何可靠性杀死Python线程要好。

就是说，该模块中有一些代码或多或少地回答了您的问题。它基本上由一个类组成，该类的对象可以设置为从优先级队列中获取位置，并将其输入到对象实例化时提供的fetch函数中。如果在杀死线程之前成功接收到该位置的资源，则将它们转发到results队列；否则，它们将以降级的优先级返回到locations队列。成功由默认为bool的传入函数确定。

在此过程中，我最终创建了terminable_thread模块，该模块只打包了我可以找到的最成熟的变体，即您链接到的代码为InterruptableThread。它还为64位计算机添加了一个修复程序，我需要此修复程序才能在ubuntu机器上使用该代码。 terminable_thread是pqueue_fetcher的依赖项。

我碰到的最大绊脚石可能是引发一个异步异常，就像terminable_thread和您提到的InterruptableThread一样，可能会产生一些奇怪的结果。在pqueue_fetcher的测试套件中，fetch函数通过调用time.sleep进行阻止。我发现如果在阻塞时线程是terminate() d，并且sleep调用是嵌套try块中的最后一个(甚至不是最后一个)语句，则执行实际上会反弹到外 try块的except子句，即使内部对象的except与引发的异常匹配也是如此。我仍然难以置信地摇了摇头，但是pqueue_fetcher中有一个测试用例可以重现这一点。我相信“泄漏抽象”是这里的正确术语。

我写了一个骇人的解决方法，它只是做一些随机的事情(在这种情况下，是从生成器中获取一个值)，以破坏该部分代码的“原子性”(不确定是否真的是这样)。可以通过fission的pqueue_fetcher.Fetcher参数覆盖此替代方法。它(即默认值)似乎可以工作，但是我绝对不会认为它特别可靠或可移植。

因此，在发现了这一有趣的数据之后，我的电话是迄今为止完全避免使用这种技术(即，调用ctypes.pythonapi.PyThreadState_SetAsyncExc)。

在任何情况下，如果您需要保证已接收到整个数据集(即已确认到服务器)的所有请求都转发到results，则此方法仍然行不通。为了确保这一点，您必须确保完成最后一次网络事务和转发的位不会被中断，而又不会阻止整个检索操作被中断(因为这样做会阻止超时。) 。为了做到这一点，您需要基本上重写检索操作(即套接字代码)，以了解要使用terminable_thread.Thread.raise_exc引发的任何异常。

我还没有学过任何东西，但是作为Premier Python异步网络框架©™®，我希望它必须具有某种优雅的方式，或者至少是可行的方式来处理这些细节。我希望它提供一种并行方式来实现从非网络源(例如本地文件存储，数据库或其他)的访存，因为我想构建一个可以从各种网络中收集数据的应用程序来源与媒介无关。

无论如何，如果您仍想尝试自己开发一种方法来管理线程，那么您也许可以向我学习。希望这可以帮助。

··········这只是在:

我已经意识到，我认为已经稳定的测试实际上并没有，并且给出了不一致的结果。这似乎与上面提到的异常处理和fission函数的使用有关。我不是很确定这是怎么回事，也不要打算在不久的将来进行调查，除非最终我需要以这种方式实际做事。