Блог

Ну, я делаю небольшую программу, которая предназначена для получения сообщений на rabbitmq и отправки клиентам, подключенным к websocket с тем же индексом, но при отправке клиенту возникла проблема, я повторно использовал код, который должен был работать в версии 3.6 . 9, и я не помню версию tornado (библиотека websocket, которую я использую), но я сменил компьютер и могу установить его снова, теперь у меня самые новые версии библиотеки и python.

Я опубликую свой старый код, потому что его легче понять из-за ошибки msm

 import tornado.websocket
import tornado.ioloop
import threading
import pika
import json

def verificar_novo(se):
    for i in range(0,len(conexao_lista)):
        if se == conexao_lista[i]["endereco"]:
            return 0
    return 1

def excluir_conexao(endereco):
    for i in range(0,len(conexao_lista)):
        if conexao_lista[i]["endereco"] == endereco:
            del(conexao_lista[i])
            break

""" Função para procurar mensagens no rabbit e retornar para os clientes"""
def callback(ch, method, properties, body):
    menssagem_rabbit = json.loads(body)
    threading.Lock()
    for i in range(0, len(conexao_lista)):
        if (conexao_lista[i]["configuracao"]["ras_eve_id_indice"]) == (menssagem_rabbit["ras_eve_id_indice"]):
            conexao_lista[i]["endereco"].write_message(menssagem_rabbit)
            break
    threading.RLock()



""" Classe de conexao com cliente"""
class WebSocketHandler(tornado.websocket.WebSocketHandler):
    def open(self):
        print("Novo cliente conectado")


    def on_close(self):
        print("Cliente desconectado")
        excluir_conexao(self)

    def on_message(self, message):

        n = verificar_novo(self)
        if n == 0:
            self.write_message(u"Sua menssagem: "   message)
        else:
            dados_json = json.loads(message)
            conexao_dicionario["endereco"] = self
            conexao_dicionario["configuracao"] = dados_json
            conexao_lista.append(conexao_dicionario.copy())
            self.write_message(u"Usuario conectado "   dados_json["id_usuario"])


    def check_origin(self, origin):
        return True

"""Função que a thread ficara rodando para consumir as mensagem em segundo plano"""
def procurar_mensagens():
    connection = pika.BlockingConnection(pika.ConnectionParameters(host='localhost'))
    channel = connection.channel()
    channel.basic_consume(queue='testerenan', on_message_callback=callback, auto_ack=True)
    channel.start_consuming()

"""Variaveis"""
conexao_lista = []
conexao_dicionario = {"endereco": "","configuracao":""}


"""Chamando a Thread"""
threading.Thread(target=procurar_mensagens, args=()).start()

"""Conexão do WebSocket"""
application = tornado.web.Application([(r"/", WebSocketHandler),])


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

Ошибка, которая появляется:

 Exception in thread Thread-1: 
Traceback (most recent call last):
  File "/usr/lib/python3.8/threading.py", line 932, in _bootstrap_inner
    self.run() 
  File "/usr/lib/python3.8/threading.py", line 870, in run
    self._target(*self._args, **self._kwargs) 
  File "/home/renan/Área de Trabalho/Projeto-WebSocket/Servidor.py", line 63, in procurar_mensagens
    channel.start_consuming()
  File "/usr/local/lib/python3.8/dist-packages/pika/adapters/blocking_connection.py", line 1866, in start_consuming
    self._process_data_events(time_limit=None) 
  File "/usr/local/lib/python3.8/dist-packages/pika/adapters/blocking_connection.py", line 2027, in _process_data_events
    self.connection.process_data_events(time_limit=time_limit)
  File "/usr/local/lib/python3.8/dist-packages/pika/adapters/blocking_connection.py", line 834, in process_data_events
    self._dispatch_channel_events()
  File "/usr/local/lib/python3.8/dist-packages/pika/adapters/blocking_connection.py", line 566, in _dispatch_channel_events
    impl_channel._get_cookie()._dispatch_events() 
  File "/usr/local/lib/python3.8/dist-packages/pika/adapters/blocking_connection.py", line 1493, in _dispatch_events
    consumer_info.on_message_callback(self, evt.method, 
  File "/home/renan/Área de Trabalho/Projeto-WebSocket/Servidor.py", line 26, in callback
    conexao_lista[i]["endereco"].write_message(menssagem_rabbit) 
  File "/home/renan/.local/lib/python3.8/site-packages/tornado/websocket.py", line 342, in write_message
    return self.ws_connection.write_message(message, binary=binary)
  File "/home/renan/.local/lib/python3.8/site-packages/tornado/websocket.py", line 1098, in write_message
    fut = self._write_frame(True, opcode, message, flags=flags) 
  File "/home/renan/.local/lib/python3.8/site-packages/tornado/websocket.py", line 1075, in _write_frame
    return self.stream.write(frame)
  File "/home/renan/.local/lib/python3.8/site-packages/tornado/iostream.py", line 555, in write
    future = Future() # type: Future[None] 
  File "/usr/lib/python3.8/asyncio/events.py", line 639, in get_event_loop
    raise RuntimeError('There is no current event loop in thread %r.' 
RuntimeError: There is no current event loop in thread 'Thread-1'.
  

Я оставлю проект для загрузки, если кто-нибудь захочет взглянуть:
https://github.com/Renan-Sacca/Projeto-WebSocket

В общем, вы должны быть очень осторожны при смешивании потоков и Tornado — вы не можете вызывать большинство методов tornado из других потоков (это всегда было правдой, но библиотека стала более строгой в применении этого в Tornado 5.0). В частности, это включает write_message . Таким образом, в callback вместо вызова write_message вы должны попросить IOLoop вызвать его для вас.

В главном блоке сделайте global main_io_loop; main_io_loop = IOLoop.current() , чтобы сохранить IOLoop основных потоков, чтобы вы могли обратиться к нему позже. Затем в callback замените вызов на write_message на

 main_io_loop.add_callback(conexao_lista[i]["endereco"].write_message, menssagem_rabbit)