如何解决如何在另一个无限循环线程中与生产者一起正确管理Websocket服务器
我正在努力解决一个容易解决但难以解决的问题。 我想设置一个websocket服务器,将消息广播到每个客户端。消息来自另一个API,该API需要进行事件的无限循环。我决定将这个无限循环放在另一个线程中,但是我很难管理线程之间的交互作用。
我确实需要主线程能够控制 Producer 线程,例如更改“客户端”列表或customMessage dict。由于无限循环将在run()中运行,因此我需要确保所有内容都是线程安全的。
这是我的代码。我在while循环中用创建随机消息以及基于客户端的自定义消息代替了API事件处理。我们等待1秒以模仿两次事件之间的时间。
from SimpleWebSocketServer import SimpleWebSocketServer,WebSocket
from random import random
import time
import threading
class Producer(threading.Thread):
def __init__(self):
self.clients = []
self.customMessage = {}
self.lock = threading.RLock()
self.session = self.connect()
super(Producer,self).__init__()
def run(self):
while True:
# ultimately this part will process received events from another api
message = f"random_number : {random()}"
for client in self.clients:
# send random number to everybody
client.sendMessage(message)
# send a custom message
if client in self.customMessage:
client.sendMessage(self.customMessage[client])
time.sleep(1)
def addCLient(self,client):
with lock:
self.clients.append(client)
def removeCLient(self,client):
with lock:
self.clients.remove(client)
def manageCustomMessage(self,client,message):
with lock:
self.customMessage[client] = message
class Consumer(WebSocket):
def handleMessage(self):
producer.manageCustomMessage(self,self.data)
def handleConnected(self):
print(self.address,'connected')
producer.addCLient(self)
def handleClose(self):
producer.removeCLient(self)
print(self.address,'closed')
producer = Producer()
producer.setDaemon(True)
producer.start()
server = SimpleWebSocketServer('127.0.0.1',8080,Consumer)
server.serveforever()
我的问题是:以我的方式使用锁是否可以?添加除运行到类线程之外的方法是否还好?
我想避免排队,因为我将需要从Producer类中获得许多属性,而我必须从主类中更改这些属性。
预先感谢
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。