Tornado PeriodicCallback 在使用较大的回调时间（即：200 毫秒）时不起作用答案

【问题标题】：Tornado PeriodicCallback does not works when using larger callback_time (ie: 200ms)Tornado PeriodicCallback 在使用较大的回调时间（即：200 毫秒）时不起作用
【发布时间】：2020-05-23 05:18:58
【问题描述】：

我有一个 websocket 服务器应用程序，它使用 Tornado PeriodicCallback 向每个 websocket 客户端发送消息。

ioloop.PeriodicCallback(dispatch, 10).start()
ioloop.IOLoop.instance().start()

dispatch() 函数有一个循环来消费 RabbitMQ 消息，然后将它们转发到每个 websocket 客户端。

def dispatch():

    global channel, queue
    time_start = time.time()
    while True:
        try:
            method_frame, header_frame, body = channel.basic_get(queue)
            if method_frame:
                message = json.loads(body.decode('utf-8'))
                if 'websocket_uri' in message:
                    websocket_uri = message['websocket_uri']
                    uri = urlparse(websocket_uri)
                    path = uri.path
                else:
                    path = ''
                if 'payload' in message:
                    payload = json.dumps(message['payload'])
                else:
                    payload = ''
                for client in clients:
                    if client.path == path:
                        client.write_message(payload)
                        logger.info('WRITE: %s: %s' % (client.path, payload))
                channel.basic_ack(method_frame.delivery_tag)
        except Exception as e:
            logger.exception(str(e))
            channel.basic_nack(method_frame.delivery_tag)
        finally:
            time_end = time.time()

        if time_end - time_start > 1:
            break;

    return

不知何故，当我使用更大的回调时间值（如 100 毫秒或 200 毫秒）时，并非所有消息都转发到 websocket 客户端。但是当我使用较小的值（如 10ms 或 1ms）时，该功能可以正常工作。

PeriodicCallback 的实际工作原理是什么？如何确保dispatch()函数总是被Tornado调用？

谢谢

【问题讨论】：

能否在dispatch() 中插入一些打印语句来检查它是否被调用？
我做到了。经过几个小时的测试和观察日志，我找到了解决方案。我只需要在dispatch函数的最后加上：ioloop.IOLoop.current().add_callback(dispatch)，并用同一行替换PeriodicCallback即可。

标签： python python-3.x websocket rabbitmq tornado

【解决方案1】：

我找到了解决方案。我用add_callback替换了PeriodicCallback：

app.listen(9001)
mainloop = ioloop.IOLoop.current()
mainloop.add_callback(dispatch)
mainloop.start()

然后在dispatch() 函数的末尾使用add_callback，这样dispatch() 函数将在下一次I/O 迭代中被调用。

def dispatch():

    global channel, queue
    while True:
        try:
            method_frame, header_frame, body = channel.basic_get(queue)
            if method_frame:
                message = json.loads(body.decode('utf-8'))
                if 'websocket_uri' in message:
                    websocket_uri = message['websocket_uri']
                    uri = urlparse(websocket_uri)
                    path = uri.path
                else:
                    path = ''
                if 'payload' in message:
                    payload = json.dumps(message['payload'])
                    logger.info('Payload: %s' % payload)
                else:
                    payload = ''
                for client in clients:
                    logger.info('Path: %s' % client.path)
                    if client.path == path:
                        client.write_message(payload)
                        logger.info('WRITE: %s: %s' % (client.path, payload))
                channel.basic_ack(method_frame.delivery_tag)
            else:
                break;
        except Exception as e:
            logger.exception(str(e))
            channel.basic_nack(method_frame.delivery_tag)

    ioloop.IOLoop.current().add_callback(dispatch)
    return

【讨论】：