Redis
 sql >> база данни >  >> NoSQL >> Redis

Flask-SocketIO redis се абонирайте

Реших предаването на приложението като аргумент на класа и използването му на контекста, както е предложено от описанието на грешката, но пространството от имена също е необходимо:

class Listener(threading.Thread):
    def __init__(self, r, channels, app):
    threading.Thread.__init__(self)
    self.daemon = True
    self.redis = r
    self.pubsub = self.redis.pubsub()
    self.pubsub.psubscribe(channels)
    self.app = app

    def work(self, item):
        with app.app_context():
            if isinstance(item['data'], bytes):
                try:
                    msg = item['data'].decode('utf-8')
                    decode_msg = json.loads(msg)                
                    if decode_msg['type'] == 'UPDATE_TASK':
                        send(json.dumps({"type":"UPDATE_TASK"}), room='home', namespace='/')
                    #_send_task_message()
                except ValueError as e:
                    log.error("Error decoding msg to microservice: %s", str(e))

    def run(self):
        for item in self.pubsub.listen():
            self.work(item)

if __name__ == '__main__':

    r = redis.Redis()
    client = Listener(r, ['/bobguarana/socketio'], app)
    client.start()

    socketio.run(debug=True, app=app, port=8080)


  1. Redis
  2.   
  3. MongoDB
  4.   
  5. Memcached
  6.   
  7. HBase
  8.   
  9. CouchDB
  1. Най-добра практика за надграждане на Redis със Sentinels?

  2. Redis:NOAUTH Изисква се удостоверяване, но няма настройка на парола

  3. Асинхронно изпълнение на redis команди

  4. NodeJS - сигурно свързване с външен Redis сървър

  5. Как да получите, актуализирате всички ключове и неговите стойности от базата данни Redis в C#?