Cách thích hợp để xử lý kết nối Redis trong Tornado là gì? (Không đồng bộ - Pub/Sub)

Question

Tôi đang sử dụng Redis cùng với ứng dụng Tornado của tôi với khách hàng Bản Noi Quy Brukva, khi tôi nhìn vào các ứng dụng mẫu tại trang web Brukva họ đang thực hiện kết nối mới trên "init" phương pháp trong WebSocket

class MessagesCatcher(tornado.websocket.WebSocketHandler): 
    def __init__(self, *args, **kwargs): 
     super(MessagesCatcher, self).__init__(*args, **kwargs) 
     self.client = brukva.Client() 
     self.client.connect() 
     self.client.subscribe('test_channel') 

    def open(self): 
     self.client.listen(self.on_message) 

    def on_message(self, result): 
     self.write_message(str(result.body)) 

    def close(self): 
     self.client.unsubscribe('test_channel') 
     self.client.disconnect() 

Phạt tiền trong trường hợp websocket nhưng cách xử lý trong phương thức đăng bài Tornado RequestHandler thông thường nói hoạt động bỏ phiếu dài (mô hình xuất bản-đăng ký). Tôi đang làm cho khách hàng mới connetion trong mỗi phương pháp đăng bài của bộ xử lý cập nhật này là phương pháp tiếp cận đúng ?? Khi tôi kiểm tra tại giao diện điều khiển redis tôi thấy rằng khách hàng ngày càng tăng trong mỗi hoạt động bài mới.

Đây là mẫu mã của tôi.

c = brukva.Client(host = '127.0.0.1') 
c.connect() 

class MessageNewHandler(BaseHandler): 
    @tornado.web.authenticated 
    def post(self): 

     self.listing_id = self.get_argument("listing_id") 
     message = { 
      "id": str(uuid.uuid4()), 
      "from": str(self.get_secure_cookie("username")), 
      "body": str(self.get_argument("body")), 
     } 
     message["html"] = self.render_string("message.html", message=message) 

     if self.get_argument("next", None): 
      self.redirect(self.get_argument("next")) 
     else: 
      c.publish(self.listing_id, message) 
      logging.info("Writing message : " + json.dumps(message)) 
      self.write(json.dumps(message)) 

    class MessageUpdatesHandler(BaseHandler): 
     @tornado.web.authenticated 
     @tornado.web.asynchronous 
     def post(self): 
      self.listing_id = self.get_argument("listing_id", None) 
      self.client = brukva.Client() 
      self.client.connect() 
      self.client.subscribe(self.listing_id) 
      self.client.listen(self.on_new_messages) 

     def on_new_messages(self, messages): 
      # Closed client connection 
      if self.request.connection.stream.closed(): 
       return 
      logging.info("Getting update : " + json.dumps(messages.body)) 
      self.finish(json.dumps(messages.body)) 
      self.client.unsubscribe(self.listing_id) 


     def on_connection_close(self): 
      # unsubscribe user from channel 
      self.client.unsubscribe(self.listing_id) 
      self.client.disconnect() 

Tôi đánh giá cao nếu bạn cung cấp một số mã mẫu cho trường hợp tương tự.

Answer 1

bạn nên gộp các kết nối trong ứng dụng của mình. vì nó có vẻ như brukva không hỗ trợ điều này tự động (redis-py hỗ trợ điều này, nhưng bị chặn bởi tự nhiên nên nó không tốt với lốc xoáy), bạn cần phải viết hồ bơi kết nối của riêng bạn. Tuy nhiên,

kiểu này khá đơn giản. một cái gì đó dọc theo các dòng này (đây không phải là mã hoạt động thực):

class BrukvaPool(): 

    __conns = {} 


    def get(host, port,db): 
     ''' Get a client for host, port, db ''' 

     key = "%s:%s:%s" % (host, port, db) 

     conns = self.__conns.get(key, []) 
     if conns: 
      ret = conns.pop() 
      return ret 
     else: 
      ## Init brukva client here and connect it 

    def release(client): 
     ''' release a client at the end of a request ''' 
     key = "%s:%s:%s" % (client.connection.host, client.connection.port, client.connection.db) 
     self.__conns.setdefault(key, []).append(client) 

nó có thể phức tạp hơn một chút, nhưng đó là ý tưởng chính.

Answer 2

Một chút trễ nhưng tôi đã sử dụng tornado-redis. Nó hoạt động với ioloop cơn lốc xoáy và các tornado.gen mô-đun

Install tornadoredis

Nó có thể được cài đặt từ pip

pip install tornadoredis 

hoặc với setuptools

easy_install tornadoredis 

nhưng bạn thực sự không nên làm việc đó đi. Bạn cũng có thể sao chép kho lưu trữ và giải nén nó. Sau đó chạy

python setup.py build 
python setup.py install 

Connect để redis

Các mã sau đi trong main.py của bạn hoặc tương đương

redis_conn = tornadoredis.Client('hostname', 'port') 
redis_conn.connect() 

redis.connect được gọi là chỉ một lần. Nó là một cuộc gọi chặn, vì vậy nó nên được gọi trước khi bắt đầu ioloop chính. Cùng một đối tượng kết nối được chia sẻ giữa tất cả các trình xử lý.

Bạn có thể thêm nó vào cài đặt ứng dụng của bạn như

settings = { 
    redis = redis_conn 
} 
app = tornado.web.Application([('/.*', Handler),], 
           **settings) 

Sử dụng tornadoredis

Các kết nối có thể được sử dụng trong xử lý tại self.settings['redis'] hoặc nó có thể được thêm vào như là một tài sản của BaseHandler và lớp con lớp đó cho các trình xử lý yêu cầu khác.

class BaseHandler(tornado.web.RequestHandler): 

    @property 
    def redis(): 
     return self.settings['redis'] 

Để giao tiếp với redis, các tornado.web.asynchronous và tornado.gen.engine trang trí được sử dụng

class SomeHandler(BaseHandler): 

    @tornado.web.asynchronous 
    @tornado.gen.engine 
    def get(self): 
     foo = yield gen.Task(self.redis.get, 'foo') 
     self.render('sometemplate.html', {'foo': foo} 

Thông tin thêm

Thêm ví dụ và các tính năng khác như kết nối pooling và đường ống dẫn có thể được tìm thấy tại github repo.