2013-02-28 16 views
11

私はRedisとTornadoをどのように非同期的に使うことができるかを探そうとしています。 tornado-redisが見つかりましたが、コードにyieldを追加するだけでは不十分です。 /waitに保留中の要求があるとしながら、私はアクセスに/ URLを取得している必要TornadoとRedisを非同期で使用するにはどうすればよいですか?

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() 

の「Hello World」を得る:

は、私は、次のコードを持っています。 どうすればいいですか?

+1

.listen() ' websocketの動作例については、http://tornadogists.org/532067/をご覧ください。 –

+0

websocketは良い選択ですが、私のアプリケーションはWebSocketをサポートしていないブラウザでも動作する必要があります。私は長いポーリングを使用しています。それが私に「非同期取得」が必要な理由です。 –

+0

@HelieelsonSantosその場合、最良の方法は、購読されたチャンネル履歴のローカル状態を(別のスレッドによって供給される)維持し、その状態をただちにレスポンスに書き込み、 'get'操作を完了することです。クライアントは、最後に取得したインデックスまたは最後に取得した時間などのレコードを維持して、異なるクライアントの連続性を維持する必要があります。私は時間を取って数時間で答えを書いてみましょう。 –

答えて

5

IOループをブロックするので、メインのトルネードスレッドでRedis pub/subを使用しないでください。メインスレッドでWebクライアントからの長いポーリングを処理できますが、Redisを待機するために別のスレッドを作成する必要があります。メッセージを受信すると、ioloop.add_callback()および/またはthreading.Queueを使用してメインスレッドと通信できます。

1

わかりましたので、ここで私はGETリクエストでそれを行うだろうかの私の例です。

Iは、2つの主要なコンポーネントを追加しました:

最初はローカルリストオブジェクトに新しいメッセージを追加単純なねじのpubsubリスナーです。 また、クラスにリストアクセサを追加したので、あなたは普通のリストから読み込んでいるかのようにリスナースレッドから読み込むことができます。 WebRequestに関する限り、ローカルリストオブジェクトからデータを読み込むだけです。これは即時に戻り、完了または将来の要求が受け入れられて処理されることからの現在の要求をブロックしません。

2番目はApplicationMixinクラスです。これは、機能と属性を追加するために継承するWeb要求クラスを持つ2次オブジェクトです。この場合、要求されたチャネルのチャネルリスナーがすでに存在するかどうかをチェックし、存在しない場合はチャネルリスナを作成し、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 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クラスは今、リスナーを扱い、私が追加しました属性

# 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() 

完全なコード:パイソン> = 3の場合

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() 
+0

リストを、非ブロッキングアクセスをサポートしているキューや他のオブジェクトで簡単に置き換えることができ、以前の要求以降に受信したメッセージのみを返すことができます。ただし、各クライアントのキューを維持し、ノンブロッキング取得を使用し、 'Empty'例外を正しく処理する必要があります。 –

2

。3、私はaioredisを使用することをお勧めします。 私は以下のコードをテストしていないが、それはそのようなものでなければなりません: `のpubsubに待っている間、それはioloopをブロックしますので、Redisのパブ/サブは、` web.RequestHandler`では使用しないでください

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() 
関連する問題