使用多处理在GUI应用程序中进行异步调用

Question

我有一个GUI应用程序,需要从GUI主循环旁边的网络中获取和解析各种资源.我使用python多处理模块搜索了选项,因为这些获取操作不仅包含阻塞IO而且还包含大量解析,因此多处理可能比python线程更好.使用Twisted会很容易,但这次Twisted不是一个选项.

我找到了一个简单的解决方案:

Python子进程:cmd退出时的回调

问题是在MainThread中没有调用回调.

所以我想出了以下解决方案:

delegate.py

import os
import multiprocessing as mp
import signal
from collections import namedtuple
import uuid
import logging


_CALLBACKS = {}
_QUEUE = mp.Queue()

info = logging.getLogger(__name__).info


class Call(namedtuple('Call', 'id finished result error')):

    def attach(self, func):
        if not self.finished:
            _CALLBACKS.setdefault(self.id, []).append(func)
        else:
            func(self.result or self.error)

        return self

    def callback(self):
        assert self.finished, 'Call not finished yet'
        r = self.result or self.error
        for func in _CALLBACKS.pop(self.id, []):
            func(r)

    def done(self, result=None, error=None):
        assert not self.finished, 'Call already finished'
        return self._replace(finished=(-1 if error else 1),
            result=result, error=error)

    @classmethod
    def create(clss):
        call = clss(uuid.uuid4().hex, 0, None, None) # uuid ???
        return call

def run(q, cb, func, args=None, kwargs=None):
    info('run: try running %s' % func)
    try:
        cb = cb.done(result=func(*(args or ()), **(kwargs or {})))
    except Exception, err:
        cb = cb.done(error=err)
    q.put(cb)
    os.kill(os.getppid(), signal.SIGUSR2) # SIGUSR2 ???
    info('run: leaving')

def on_callback(sig, frame):
    info('on_callback: checking queue ...')
    c = _QUEUE.get(True, 2)
    info('on_callback: got call - %s' % repr(c))
    c.callback()

signal.signal(signal.SIGUSR2, on_callback) # SIGUSR2 ???

def delegate(func, *args, **kwargs):
    info('delegate: %s %s' % (func, args,))
    cb = Call.create()
    mp.Process(target=run, args=(_QUEUE, cb, func, args, kwargs,)).start()
    return cb


__all__ = ['delegate']

用法

from delegate import delegate

def sleeper(secs):
    assert secs >= 1, 'I need my Augenpflege'
    info('sleeper: will go to sleep for %s secs' % secs)
    sleep(secs)
    info('sleeper: woke up - returning result')
    return ['sleeper', 'result']

def on_sleeper_result(r):
    if isinstance(r, Exception):
        info('on_sleeper_result: got error: %s' % r)
    else:
        info('on_sleeper_result: got result: %s' % r)

from delegate import delegate
delegate(sleeper, 3).attach(on_sleeper_result)
delegate(sleeper, -3).attach(on_sleeper_result)
while 1:
    info('main: loop')
    sleep(1)

产量

0122 08432 MainThread INFO  delegate: <function sleeper at 0x163e320> (3,)
0123 08432 MainThread INFO  delegate: <function sleeper at 0x163e320> (-3,)
0124 08437 MainThread INFO  run: try running <function sleeper at 0x163e320>
0124 08437 MainThread INFO  sleeper: will go to sleep for 3 secs
0124 08432 MainThread INFO  main: loop
0125 08438 MainThread INFO  run: try running <function sleeper at 0x163e320>
0126 08438 MainThread INFO  run: leaving
0126 08432 MainThread INFO  on_callback: checking queue ...
0126 08432 MainThread INFO  on_callback: got call - Call(id='057649cba7d840e3825aa5ac73248f78', finished=-1, result=None, error=AssertionError('I need my Augenpflege',))
0127 08432 MainThread INFO  on_sleeper_result: got error: I need my Augenpflege
0127 08432 MainThread INFO  main: loop
1128 08432 MainThread INFO  main: loop
2129 08432 MainThread INFO  main: loop
3127 08437 MainThread INFO  sleeper: woke up - returning result
3128 08437 MainThread INFO  run: leaving
3128 08432 MainThread INFO  on_callback: checking queue ...
3129 08432 MainThread INFO  on_callback: got call - Call(id='041420c6c83a489aa5c7409c662d4917', finished=1, result=['sleeper', 'result'], error=None)
3129 08432 MainThread INFO  on_sleeper_result: got result: ['sleeper', 'result']
3129 08432 MainThread INFO  main: loop
4130 08432 MainThread INFO  main: loop
5132 08432 MainThread INFO  main: loop
...

到目前为止,这种方法运行良好,但我对多处理模块的体验是适度的,我有点不确定这是否会在没有效果的情况下运行.我的问题是 - 在以这种方式使用多处理时我应该特别关注的是什么？或者使用python标准库的异步回调机制是否有'更正确'的模式？

Answer 1

您没有理由使用信号（低级 api）进行 python 多处理并忙于在主循环中等待。

您必须在 a 中运行（修改后的）事件循环QThread，它可以直接调用 qt 代码，或者使用 QApplication.postEvent （或 pyqtSignal）在主线程中执行它

# this should be in the delegate module
while 1:
    c = _QUEUE.get(True) # no timeout
    c.callback() # or post event to main thread

您还可以查看此页面有关 qt 中线程之间通信的讨论