本地套接字和 websocket 占用 CPU 的 Python“代理”

Question

我当前的项目是安装在 din 导轨上的 Raspberry pi 4，带有两个额外的电路板。 它应该从各种来源收集数据，例如。 CAN、RS485、GPIO 引脚、对来自电能/水/气表等的 S0 脉冲进行计数，将计数值存储到 eeram 中并在网页上实时显示。

对于每个源，都有一个 python 脚本（作为守护程序运行）收集数据并在必要时（在更改时）将其存储/传输到在端口 10000 上打开的本地套接字。

主“代理”守护进程监听端口 10000，并将接收到的数据转发到端口 9999 上的 websocket 客户端（1 个或多个）。 然后使用 JavaScript 将数据插入到网页中，如果服务器不可用，JavaScript 会自动重新连接。

要与系统交互，可以按下按钮或输入作为命令通过“代理守护程序”发送的文本。守护进程立即将命令分发给连接的本地套接字客户端。

守护进程简化了三个任务：

• 监听 localhost 端口 10000，接受多个客户端
• 监听 websocket 端口 9999，接受多个客户端
• 将接收到的数据双向转发到所有连接的客户端

我已经阅读了许多关于线程、套接字、异步、事件循环的资料，但是如果连接了本地套接字，我已经在工作的脚本会占用 cpu。这是因为第 67 行中的循环。 有没有更好的方法来实现目标，最好不用线程？

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import time, threading, socket, signal, selectors, asyncio, types, urllib
import tornado.httpserver, tornado.websocket, tornado.ioloop, tornado.web

wsclients = []
host = 'localhost'
port = 10000
websocket_port = 9999
commands = []
sel = 0
lsock = 0

class WSHandler(tornado.websocket.WebSocketHandler):
    def open(self):
        wsclients.append(self)
        print("[WS] New websocket connection from", self.request.remote_ip)

    def check_origin(self, origin):  
        parsed_origin = urllib.parse.urlparse(origin)
        print("[WS] origin:", origin)
        return parsed_origin.netloc.endswith("raspi")
        
    def on_message(self, message):
        print("[WS] Received:", message)
        # Forward message to localhost socket clients
        commands.append(message)

    def on_close(self):
        wsclients.remove(self)
        print("[WS] Closed connection from", self.request.remote_ip)


def broadcast(message):
    global wsclients
    for client in wsclients:
        client.write_message(message)
    

def init_socket():
    global sel
    global lsock

    sel = selectors.DefaultSelector()
    lsock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    lsock.bind((host, port))
    lsock.listen()
    print("[LS] Server listening on %s %d" %(host, port))
    lsock.setblocking(False)
    sel.register(lsock, selectors.EVENT_READ, data=None)
    
    t = threading.Thread(target=socket_loop)
    t.daemon=True
    t.start()

def socket_loop():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    loop.create_task(socket_event())
    loop.run_forever()
    loop.stop()

async def socket_event():
    while True:
        events = sel.select(timeout=None)
        for key, mask in events:
            if key.data is None:
                accept_wrapper(key.fileobj)
            else:
                service_connection(key, mask)

def accept_wrapper(sock):
    conn, addr = sock.accept()
    print("[LS] Connected with " + addr[0] + ":" + str(addr[1]))
    conn.setblocking(False)
    data = types.SimpleNamespace(addr=addr, inb=b'', outb=b'', port=addr[1])
    sel.register(conn, selectors.EVENT_READ | selectors.EVENT_WRITE, data=data)

def service_connection(key, mask):
    sock = key.fileobj
    data = key.data
    if mask & selectors.EVENT_READ:
        recv_data = sock.recv(1024)
        if recv_data:
            print("[LS] Received:", recv_data)
            # Forward message to websocket clients
            broadcast(str(recv_data.decode()) + "\n")
        else:
            print('[LS] closing connection to', data.addr)
            sel.unregister(sock)
            sock.close()
    if mask & selectors.EVENT_WRITE:
        if not data.outb and commands:
            data.outb = commands.pop(0).encode()
        if data.outb:
            print('[LS] sending', repr(data.outb), 'to connection port', data.port)
            sent = sock.send(data.outb)
            data.outb = data.outb[sent:]


async def shutdown():
    for client in wsclients:
        client.close()
    http_server.stop()
    lsock.close()
    tornado.ioloop.IOLoop.current().stop()
    print("[*] Exiting..")

def exit_handler(sig, frame):
#    print("[*] Signal:" + str(sig))
    tornado.ioloop.IOLoop.instance().add_callback_from_signal(shutdown)

app = tornado.web.Application([(r'/', WSHandler),])

if __name__ == "__main__":
    signal.signal(signal.SIGTERM, exit_handler)
    signal.signal(signal.SIGINT,  exit_handler)

    init_socket()
    
    http_server = tornado.httpserver.HTTPServer(app)
    http_server.listen(websocket_port)
    print("[WS] Webserver listening on %d" %(websocket_port))

    tornado.ioloop.IOLoop.instance().start()
    #    t.join()

Answer 1

您遇到的主要问题是，当您在输出缓冲区中有数据要发送时，您只想注册selectors.EVENT_WRITE。否则选择器会（正确地）告诉您每次循环都可以写入套接字，但service_connection 将无事可做。您需要根据输出缓冲区的状态更新您的注册。

您还将许多并非真正必要的东西组合在一起。 selectors 模块是您用来实现自己的事件循环的那种东西，因此从另一个事件循环中运行它是不正确的（但因为您没有在 asyncio 循环中运行其他任何东西你的线程在这种情况下是无害的）。只选择以下四个选项之一来实现您的服务器：Tornado、异步、带选择器的线程或不带选择器的线程（阻塞）。

我建议您使用 Tornado，因为这就是您用于 websockets 的东西。根据您对低级套接字编程的喜爱程度，您有几个选择。您可以使用IOLoop.add_handler 编写一些与service_connection 中的代码非常相似的代码，或者您可以使用IOStream 来拥有更高级别的接口。