0

我正在使用 ZMQ 流，它们基于他们自己的龙卷风 IOLoop 实现。我遇到了一个我想要同步行为的问题，但 IOLoop 迫使我进入异步执行。以下是具体情况：

我有一个单独负责相机设置的类（相机处于不同的进程中）：

class HamSettingsManager(object):
    ''' 
        This class makes sure the camera is set to the requested settings.
    '''
    def __init__(self, blabla, loop=None):
        self._msg = messaging.CamSettings(blabla)

        self._loop = loop or zmq.eventloop.ioloop.IOLoop()
        self._sub = messaging.StreamFactory.sub_stream(
            messaging.channel_map.DI().settings,
            loop=self._loop,
            on_recv=self._on_settings)

        self._pub = messaging.StreamFactory.pub_stream(
            messaging.channel_map.DI().update_settings,
            loop=self._loop)

    @tornado.gen.coroutine
    def set_settings(self):
        self._pub.send(self._msg.SerializeToString())
        self._new_settings = True
        self._changed = None
        while self._new_settings:
            log.debug("Waiting for camera settings to take effect.")
            yield tornado.gen.Task(self._loop.add_timeout, time.time() + 0.05)

    def _on_settings(self, data):
        msg = messaging.HamSettings()
        msg.ParseFromString(data[-1])

        if msg == self._msg:
            self._new_settings = False
        else:
            if not self._new_settings:
                log.warning("Someone has changed the camera settings underneath the aligner.")
                self._changed = time.time()

            log.debug("Settings not as expected, resending")
            log.debug("Current settings: \n%s" % msg)
            log.debug("Expected settings: \n%s" % self._msg)
            self._pub.send(self._msg.SerializeToString())

我期望发生的是 set_settings 只会在设置生效后返回，但实际发生的是执行只是继续，就像我产生了一个线程一样。在 IOLoop 中等待的正确方法是什么？

4

1 回答 1

2

你不能在不停止事件循环的情况下创建阻塞调用；这就是“阻塞”的意思。作为协程，立即set_settings返回一个Future；调用者负责等待Future解决。一般来说，这意味着调用它的任何东西也必须是协程（或其他异步）并且使用yield mgr.set_settings()而不是简单调用。

于 2014-01-09T17:58:20.867 回答