-->

如何使用龙卷风和Redis的异步？(How can I use Tornado and Redis

2019-08-17 05:37发布

站内文章 / 移动开发
25 0 0

我试图找到我如何使用Redis的和龙卷风是异步的。 我发现龙卷风redis的 ，但我需要的不仅仅是增加yield中的代码。

我有以下代码： 

import redis
import tornado.web

class WaiterHandler(tornado.web.RequestHandler):

    @tornado.web.asynchronous
    def get(self):
        client = redis.StrictRedis(port=6279)
        pubsub = client.pubsub()
        pubsub.subscribe('test_channel')

        for item in pubsub.listen():
            if item['type'] == 'message':
                print item['channel']
                print item['data']

        self.write(item['data'])
        self.finish()


class GetHandler(tornado.web.RequestHandler):

    def get(self):
        self.write("Hello world")


application = tornado.web.Application([
    (r"/", GetHandler),
    (r"/wait", WaiterHandler),
])

if __name__ == '__main__':
    application.listen(8888)
    print 'running'
    tornado.ioloop.IOLoop.instance().start()

我需要获得访问/ URL，并得到了“Hello World”的同时，有一个在挂起的请求/wait 。 我该怎么做？

Answer 1:

你不应该在主线程龙卷风Redis的使用发布/订阅，因为它会阻止IO循环。 您可以处理来自Web客户端的长轮询在主线程，但你应该创造适合听Redis的一个单独的线程。 然后，您可以使用ioloop.add_callback()和/或threading.Queue当您收到消息，与主线程通信。



Answer 2:

您需要使用龙卷风IOLoop兼容Redis的客户端。

有几个人可用， toredis ， brukva等。

下面是toredis发布订阅例如： https://github.com/mrjoes/toredis/blob/master/tests/test_handler.py



Answer 3:

对于Python> = 3.3，我会建议你使用aioredis 。 我没有测试下面的代码，但它应该是这样的： 

import redis
import tornado.web
from tornado.web import RequestHandler

import aioredis
import asyncio
from aioredis.pubsub import Receiver


class WaiterHandler(tornado.web.RequestHandler):

    @tornado.web.asynchronous
    def get(self):
        client = await aioredis.create_redis((host, 6279), encoding="utf-8", loop=IOLoop.instance().asyncio_loop)

        ch = redis.channels['test_channel']
        result = None
        while await ch.wait_message():
            item = await ch.get()
            if item['type'] == 'message':
                print item['channel']
                print item['data']
                result = item['data']

        self.write(result)
        self.finish()


class GetHandler(tornado.web.RequestHandler):

    def get(self):
        self.write("Hello world")


application = tornado.web.Application([
    (r"/", GetHandler),
    (r"/wait", WaiterHandler),
])

if __name__ == '__main__':
    print 'running'
    tornado.ioloop.IOLoop.configure('tornado.platform.asyncio.AsyncIOLoop')
    server = tornado.httpserver.HTTPServer(application)
    server.bind(8888)
    # zero means creating as many processes as there are cores.
    server.start(0)
    tornado.ioloop.IOLoop.instance().start()


Answer 4:

好了，这是我的我怎么会用GET请求做例子。

我加了两个主要组件：

第一种是简单的螺纹发布订阅监听器，追加新的消息到本地列表对象。 我还添加列表存取到类，这样你就可以从监听线程，如果你是从常规列表读取读取。 至于你WebRequest而言，你只是从本地列表对象读取数据。 这立即返回，并不会阻止从完成或不被接受和处理将来的请求当前的请求。 

class OpenChannel(threading.Thread):
    def __init__(self, channel, host = None, port = None):
        threading.Thread.__init__(self)
        self.lock = threading.Lock()
        self.redis = redis.StrictRedis(host = host or 'localhost', port = port or 6379)
        self.pubsub = self.redis.pubsub()
        self.pubsub.subscribe(channel)

        self.output = []

    # lets implement basic getter methods on self.output, so you can access it like a regular list
    def __getitem__(self, item):
        with self.lock:
            return self.output[item]

    def __getslice__(self, start, stop = None, step = None):
        with self.lock:
            return self.output[start:stop:step]

    def __str__(self):
        with self.lock:
            return self.output.__str__()

    # thread loop
    def run(self):
        for message in self.pubsub.listen():
            with self.lock:
                self.output.append(message['data'])

    def stop(self):
        self._Thread__stop()

第二个是ApplicationMixin类。 这你有你的Web请求类的辅助对象继承，以增加功能和属性。 在这种情况下，它会检查通道侦听器是否已经存在请求的渠道，创建一个如果没有被发现，并返回监听器处理到的WebRequest。 

# add a method to the application that will return existing channels
# or create non-existing ones and then return them
class ApplicationMixin(object):
    def GetChannel(self, channel, host = None, port = None):
        if channel not in self.application.channels:
            self.application.channels[channel] = OpenChannel(channel, host, port)
            self.application.channels[channel].start()
        return self.application.channels[channel]

WebRequest类现在把听者仿佛它是一个静态列表（记住，你需要给self.write字符串） 

class ReadChannel(tornado.web.RequestHandler, ApplicationMixin):
    @tornado.web.asynchronous
    def get(self, channel):
        # get the channel
        channel = self.GetChannel(channel)
        # write out its entire contents as a list
        self.write('{}'.format(channel[:]))
        self.finish() # not necessary?

最后，创建应用程序后，我添加了一个空的字典作为一个属性

# add a dictionary containing channels to your application
application.channels = {}

除了正在运行的线程的一些清理，一旦你退出应用程序

# clean up the subscribed channels
for channel in application.channels:
    application.channels[channel].stop()
    application.channels[channel].join()

完整的代码： 

import threading
import redis
import tornado.web



class OpenChannel(threading.Thread):
    def __init__(self, channel, host = None, port = None):
        threading.Thread.__init__(self)
        self.lock = threading.Lock()
        self.redis = redis.StrictRedis(host = host or 'localhost', port = port or 6379)
        self.pubsub = self.redis.pubsub()
        self.pubsub.subscribe(channel)

        self.output = []

    # lets implement basic getter methods on self.output, so you can access it like a regular list
    def __getitem__(self, item):
        with self.lock:
            return self.output[item]

    def __getslice__(self, start, stop = None, step = None):
        with self.lock:
            return self.output[start:stop:step]

    def __str__(self):
        with self.lock:
            return self.output.__str__()

    # thread loop
    def run(self):
        for message in self.pubsub.listen():
            with self.lock:
                self.output.append(message['data'])

    def stop(self):
        self._Thread__stop()


# add a method to the application that will return existing channels
# or create non-existing ones and then return them
class ApplicationMixin(object):
    def GetChannel(self, channel, host = None, port = None):
        if channel not in self.application.channels:
            self.application.channels[channel] = OpenChannel(channel, host, port)
            self.application.channels[channel].start()
        return self.application.channels[channel]

class ReadChannel(tornado.web.RequestHandler, ApplicationMixin):
    @tornado.web.asynchronous
    def get(self, channel):
        # get the channel
        channel = self.GetChannel(channel)
        # write out its entire contents as a list
        self.write('{}'.format(channel[:]))
        self.finish() # not necessary?


class GetHandler(tornado.web.RequestHandler):

    def get(self):
        self.write("Hello world")


application = tornado.web.Application([
    (r"/", GetHandler),
    (r"/channel/(?P<channel>\S+)", ReadChannel),
])


# add a dictionary containing channels to your application
application.channels = {}


if __name__ == '__main__':
    application.listen(8888)
    print 'running'
    try:
        tornado.ioloop.IOLoop.instance().start()
    except KeyboardInterrupt:
        pass

    # clean up the subscribed channels
    for channel in application.channels:
        application.channels[channel].stop()
        application.channels[channel].join()


文章来源: How can I use Tornado and Redis asynchronously?
标签： python asynchronous redis tornado
举报

收藏的人(0)

Ta的文章 更多文章
登录 后发表评论
0条评论
还没有人评论过~