如何在 Web 套接字中从服务器向客户端发送消息
How to send a message from the server to the client in a web socket
好吧,我正在做一个小程序,旨在在 rabbitmq 上接收消息并发送到连接到具有相同索引的 websocket 的客户端,但是在发送到客户端时遇到问题,我重新使用了一个代码在 3.6 .9 中工作,我不记得龙卷风的版本(我使用的 websocket 库),但我更换了电脑,我能够再次安装它,现在我有了最新版本的库和 python.
我会post我的旧代码,因为它更容易理解,因为 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
中,你必须让 IOLoop
为你调用它,而不是调用 write_message
。
在主程序块中,执行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)
好吧,我正在做一个小程序,旨在在 rabbitmq 上接收消息并发送到连接到具有相同索引的 websocket 的客户端,但是在发送到客户端时遇到问题,我重新使用了一个代码在 3.6 .9 中工作,我不记得龙卷风的版本(我使用的 websocket 库),但我更换了电脑,我能够再次安装它,现在我有了最新版本的库和 python.
我会post我的旧代码,因为它更容易理解,因为 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
中,你必须让 IOLoop
为你调用它,而不是调用 write_message
。
在主程序块中,执行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)