python - Python Tornado-如何实现长轮询服务器以从队列读取

我正在尝试构建Web服务器以通过AJAX收集“命令”，然后通过长轮询将命令分发给客户端。

目标是有人将一些数据发布到/ add-command。

另一个客户端实现了一个长时间轮询的客户端，该客户端按/ poll等待命令执行。

我认为队列是用于保留命令以等待关注的正确数据结构。我希望这些命令本质上可以立即分发给任何长期轮询的客户端，但是如果当前没有客户端正在轮询，则可以保留该命令。

这是我的python脚本。

import os
import time
import tornado.httpserver
import tornado.ioloop
import tornado.web
import tornado.gen
import Queue
import multiprocessing.pool
import mysql.connector
import urlparse
import uuid
import json

_commandQueue = Queue.Queue()
_commandPollInterval = 0.2
_commandPollTimeout = 10

class HomeHandler(tornado.web.RequestHandler):
    def get(self):
        self.render("home.htm")

class AddCommandHandler(tornado.web.RequestHandler):
    def post(self):
        d = urlparse.parse_qs(self.request.body)
        _commandQueue.put(d)
        self.write(str(True))

class PollHandler(tornado.web.RequestHandler):
    @tornado.gen.coroutine
    def get(self):
        self.write("start")
        d = 1
        d = yield self.getCommand()
        self.write(str(d))
        self.write("end")
        self.finish()
    @tornado.gen.coroutine
    def getCommand(self):
        start = time.time()
        while (time.time() - start) < _commandPollTimeout * 1000:
            if not _commandQueue.empty:
                return _commandQueue.get()
            else:
                time.sleep(_commandPollInterval)
        return None

def main():
    application = tornado.web.Application(
        [
            (r"/", HomeHandler),
            (r"/add-command", AddCommandHandler),
            (r"/poll", PollHandler),
        ],
        debug=True,
        template_path=os.path.join(os.path.dirname(__file__), "templates"),
        static_path=os.path.join(os.path.dirname(__file__), "static"),
    )
    tornado.httpserver.HTTPServer(application).listen(int(os.environ.get("PORT", 5000)))
    tornado.ioloop.IOLoop.instance().start()

if __name__ == "__main__":
    main()

AddCommandHandler可以很好地将项目放入_commandQueue中。

PollHandler请求只是超时。如果我调用PollHandler，它似乎已锁定_commandQueue，而我无法放置它或从中获取它。

我怀疑我需要加入队列，但是我似乎无法在代码中找到合适的时间。

更新-这是我的最终代码，这要归功于答案

import os
import time
import datetime
import tornado.httpserver
import tornado.ioloop
import tornado.web
import tornado.gen
import tornado.queues
import urlparse
import json

_commandQueue = tornado.queues.Queue()
_commandPollInterval = 0.2
_commandPollTimeout = 10

class HomeHandler(tornado.web.RequestHandler):
    def get(self):
        self.render("home.htm")

class AddCommandHandler(tornado.web.RequestHandler):
    def get(self):
        cmd = urlparse.parse_qs(self.request.body)
        _commandQueue.put(cmd)
        self.write(str(cmd))
    def post(self):
        cmd = urlparse.parse_qs(self.request.body)
        _commandQueue.put(cmd)
        self.write(str(cmd))

class PollHandler(tornado.web.RequestHandler):
    @tornado.gen.coroutine
    def get(self):
        cmd = yield self.getCommand()
        self.write(str(cmd))
    @tornado.gen.coroutine
    def getCommand(self):
        try:
            cmd = yield _commandQueue.get(
                timeout=datetime.timedelta(seconds=_commandPollTimeout)
            )
            raise tornado.gen.Return(cmd)
        except tornado.gen.TimeoutError:
            raise tornado.gen.Return()

def main():
    application = tornado.web.Application(
        [
            (r"/", HomeHandler),
            (r"/add-command", AddCommandHandler),
            (r"/poll", PollHandler),
        ],
        debug=True,
        template_path=os.path.join(os.path.dirname(__file__), "templates"),
        static_path=os.path.join(os.path.dirname(__file__), "static"),
    )
    tornado.httpserver.HTTPServer(application).listen(int(os.environ.get("PORT", 5000)))
    tornado.ioloop.IOLoop.instance().start()

if __name__ == "__main__":
    main()

最佳答案

在异步模型中，您应该省略阻塞操作，在代码中time.sleep是有害的。此外，我认为最好的方法是使用龙卷风的（在异步接口中）队列-tornado.queue.Queue并使用async get：

import datetime
import tornado.gen
import tornado.queues

_commandQueue = tornado.queues.Queue()


    # ...rest of the code ...

    @tornado.gen.coroutine
    def getCommand(self):
        try:
            # wait for queue item if cannot obtain in timeout raise exception
            cmd = yield _commandQueue.get(
                timeout=datetime.timedelta(seconds=_commandPollTimeout)
            )
            return cmd
        except tornado.gen.Timeout:
            return None

注意：模块tornado.queues si自Tornado 4.x起可用，如果使用较旧的模块，则Toro将有所帮助。

关于python - Python Tornado-如何实现长轮询服务器以从队列读取，我们在Stack Overflow上找到一个类似的问题：https://stackoverflow.com/questions/41418332/