Python中websockets服务端从客户端接收消息并发送给多线程

这篇具有很好参考价值的文章主要介绍了Python中websockets服务端从客户端接收消息并发送给多线程。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

目录

一、消息队列

二、服务端

三、设备功能

四、主线程

五、客户端

六、更新


思路:

1.websockets需要从客户端接收消息,由于websockets创建服务端只能绑定一个端口,所以需要单独占用一个线程。收到的消息,我们需要共享给主线程,然后主线程根据设备(多线程)分发消息给各线程

2.消息中心需要独立出来,websockets服务端放消息,主线程去消息

3.根据思路设计模块:

                        1.消息库

                        2.服务端

                        3.主线程

                        4.多线程

先运行Main.py,再运行websocket_client.py(客户端),客户端发送的消息可能不一样,所以统一消息里面必须有device_id 或者device_name

修改websocket_client.py中data的信息,发送不同消息

一、消息队列

message_base.py

根据设备,创建储存设备消息,提取设备消息的功能

from queue import Queue


class MessageBase:
    def __init__(self):
        self.data = dict()

    def add(self, device, data):
        if device in self.data:
            self.data[device].put(data)
        else:
            self.data[device] = Queue()
            self.data[device].put(data)

    def get(self, device):
        data_queue: Queue = self.data.get(device)
        if not data_queue or data_queue.empty():
            return None
        data = data_queue.get()
        return data


if __name__ == '__main__':
    mb = MessageBase()
    mb.add("a", "asdasd")
    mb.add("a", "11111111")
    print(mb.data)
    data = mb.get("a")
    print(data)

二、服务端

websocket_server.py

从客户端接收消息,并存到消息队列

import asyncio
import json
import threading
import websockets
##
from message_base import MessageBase


class WebServer:
    def __init__(self, host, port, message_base: MessageBase):
        self.host = host
        self.port = port
        self.clients = []
        self.message_base = message_base

    async def echo(self, websocket, path):
        self.clients.append(websocket)
        client_ip, client_port = websocket.remote_address
        print(f"连接到:{client_ip}:{client_port}")
        while True:
            try:
                recv_text = await websocket.recv()
                data = json.loads(recv_text)
                device = data.get("device")
                if device:
                    self.message_base.add(device, data)
                else:
                    continue
            except websockets.ConnectionClosed:
                print("ConnectionClosed...")  # 链接断开
                self.clients.remove(websocket)
                break
            except websockets.InvalidState:
                print("InvalidState...")  # 无效状态
                self.clients.remove(websocket)
                break
            except Exception as e:
                print(e)

    def connect(self):
        print("连接成功!")
        asyncio.set_event_loop(asyncio.new_event_loop())
        start_server = websockets.serve(self.echo, self.host, self.port)
        asyncio.get_event_loop().run_until_complete(start_server)
        asyncio.get_event_loop().run_forever()

    def run(self):
        t = threading.Thread(target=self.connect)
        t.start()
        print("已启动!")


if __name__ == '__main__':
    mb = MessageBase()
    ws = WebServer("192.168.6.28", 8001, mb)
    ws.run()

三、设备功能

device_function.py

每个设备对应的线程功能,可以统一也可以写多个功能

class DeviceFunc:
    def __init__(self, device_name, data):
        self.device_name = device_name
        self.data = data

    def show_data(self):
        if self.data:
            print(self.device_name, "收到消息:", self.data.get("value"))

四、主线程

main.py

初始化所有功能模块,并运行主线程

from message_base import MessageBase
from websocket_server import WebServer
from device_function import DeviceFunc


class MainThread:
    def __init__(self, message_base: MessageBase, websocket_server: WebServer, device_list):
        self.message_base = message_base
        self.websocket_server = websocket_server
        self.device_list = device_list

    def run_server(self):
        self.websocket_server.run()

    def run(self):
        self.run_server()
        while True:
            for device in self.device_list:
                try:
                    # 开始根据设备即功能处理消息
                    data = self.message_base.get(device)
                    if not data:
                        continue
                    df = DeviceFunc(device, data)
                    df.show_data()
                except Exception as err:
                    pass


if __name__ == '__main__':
    mb = MessageBase()
    ws = WebServer("192.168.6.28", 8000, mb)
    device_list = ["aa", "bb", "cc"]

    mt = MainThread(mb, ws, device_list)
    mt.run()

五、客户端

webscoket_client.py

给服务端发送消息,测试用

import json

import websocket


class WebClient:
    def __init__(self, host, port):
        self.host = host
        self.port = port
        self.conn = None
        self.flag = False

    def connect(self):
        try:
            url = f"ws://{self.host}:{self.port}"
            self.conn = websocket.create_connection(url)
            self.flag = True
            print("连接成功")
        except Exception as err:
            self.flag = False
            print("连接失败", err)

    def close(self):
        self.conn.close()

    def recv(self):
        data = self.conn.recv(1024)
        print(data)

    def send(self, data):
        self.conn.send(data)
        print("发送成功")


if __name__ == '__main__':
    host = "192.168.6.28"
    port = 8000
    ws = WebClient(host, port)
    if not ws.flag:
        ws.connect()
    devices = ["aa", "bb", "cc"]
    while True:
        device = random.choice(devices)
        s = ""
        for i in range(random.randint(0, 100)):
            s += chr(random.randint(65, 122))
        data = {"device": device, "value": s}
        data = json.dumps(data)
        ws.send(data)
        time.sleep(1)

六、更新

队列在获取消息时有阻塞的现象,我们上面避免阻塞用了下面注释的内容,

   

def get(self, device):
    data_queue: Queue = self.data.get(device)
    if not data_queue:  # 阻塞
        return None
    # if not data_queue or data_queue.empty():
    #     return None
    data = data_queue.get()
    return data

我们如果不用注释的内容,让获取消息直接进入阻塞,但是阻塞是在设备线程里面,不影响主线程和其他线程。这样,我们需要把设备功能封装到线程里面。这样设备工作不会相互影响

DeviceFunc对象改为DeviceThread

import threading


class DeviceThread(threading.Thread):
    def __init__(self, device_name, message_base):
        super().__init__(target=self.process)
        self.device_name = device_name
        self.message_base = message_base

    def get_data(self):
        data = self.message_base.get(self.device_name)
        return data

    def process(self):
        while True:
            data = self.get_data()
            if data:
                print(self.device_name, "收到消息:", data.get("value"))

主线程的run方法修改为:文章来源地址https://www.toymoban.com/news/detail-857650.html

def run(self):
    self.run_server()
    # while True:
    #     for device in self.device_list:
    #         try:
    #             data = self.message_base.get(device)
    #             if not data:
    #                 continue
    #             df = DeviceFunc(device, data)
    #             df.show_data()
    #         except Exception as err:
    #             pass
    for device in self.device_list:
        t = DeviceThread(device, self.message_base)
        t.start()

到了这里,关于Python中websockets服务端从客户端接收消息并发送给多线程的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • python用websockets创建服务端websocket创建客户端

    服务端 客户端

    2024年02月22日
    浏览(55)
  • Python3实现WebSocket服务端与客户端通信

    Python3实现WebSocket服务端与客户端通信 WebSocket是一种在单个TCP连接上进行全双工通信的协议。WebSocket通信更加轻便、高效,比传统的HTTP通信更省流量和更快速,因此在Web应用领域越来越受欢迎。Python 3提供了内置的websocket库,可以方便地实现WebSocket服务端和客户端的通信。

    2024年02月12日
    浏览(53)
  • java后端使用websocket实现与客户端之间接收及发送消息

    客户端请求websocket接口,连接通道=》我这边业务成功客户端发消息=》客户端自动刷新。 接口:ws://localhost:8080/websocket/xx 经测试,成功 如果是线上服务器连接,则需要在nginx里配置websocket相关内容,再重启nginx,代码如下 本地连接的时候用的是ws://,因为是http链接,但是如果是

    2024年02月16日
    浏览(46)
  • Winform中使用Websocket4Net实现Websocket客户端并定时存储接收数据到SQLite中

    SpringBoot+Vue整合WebSocket实现前后端消息推送: SpringBoot+Vue整合WebSocket实现前后端消息推送_websocket vue3.0 springboot 往客户端推送-CSDN博客 上面实现ws推送数据流程后,需要在windows上使用ws客户端定时记录收到的数据到文件中,这里 文件使用SQLite数据库进行存储。 Winform中操作Sql

    2024年02月22日
    浏览(48)
  • SpringBoot+WebSocket实现服务端、客户端

    小编最近一直在使用springboot框架开发项目,毕竟现在很多公司都在采用此框架,之后小编也会陆续写关于springboot开发常用功能的文章。 什么场景下会要使用到websocket的呢? websocket主要功能就是实现网络通讯,比如说最经典的客服聊天窗口、您有新的消息通知,或者是项目与

    2024年02月13日
    浏览(50)
  • 当在本地,向服务器发送信息时,服务器接受信息返回给客户端,此时采用多行读取时,客户端接收不到Server的信息

        此时的服务器还在等待输入,客户端也在等待服务器的应答,产生死锁,僵持不下。 经过一番尝试后发现, 当使用单行读取的时候,使用 bufferedWriter.newLine(); 此时就不用关闭socket的输出流 , 即socket.shutdownOutput();可以省略,当然写上去也没错。 当使用多行读取的时候,

    2024年02月07日
    浏览(49)
  • C++实现WebSocket通信(服务端和客户端)

    天行健,君子以自强不息;地势坤,君子以厚德载物。 每个人都有惰性,但不断学习是好好生活的根本,共勉! 文章均为学习整理笔记,分享记录为主,如有错误请指正,共同学习进步。 这里单纯是个人总结,如需更官方更准确的websocket介绍可百度 websocket是一种即时通讯协

    2024年02月09日
    浏览(48)
  • java webSocket服务端、客户端、心跳检测优雅解决

    项目分为三个端,项目之间需要webSocket通信。 WebSocketConfig WebSocketServer

    2024年02月17日
    浏览(40)
  • Java实现WebSocket客户端和服务端(简单版)

    天行健,君子以自强不息;地势坤,君子以厚德载物。 每个人都有惰性,但不断学习是好好生活的根本,共勉! 文章均为学习整理笔记,分享记录为主,如有错误请指正,共同学习进步。 写在前面: WebSocket是一种在单个TCP连接上进行全双工通信的协议。 WebSocket通信协议于

    2024年02月08日
    浏览(48)
  • SpringBoot集成WebSocket实现客户端与服务端通信

    话不多说,直接上代码看效果! 一、服务端: 1、引用依赖 2、添加配置文件 WebSocketConfig 3、编写WebSocket服务端接收、发送功能   声明接口代码:   实现类代码: 4、如果不需要实现客户端功能,此处可选择前端调用,奉上代码 二、客户端: 1、引用依赖 2、自定义WebSocket客

    2024年01月23日
    浏览(54)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包