如何从Python文件向常驻运行的Python应用传参调用函数？

针对你的需求，这里提供几种直接高效的实现方案，替代文件扫描的方式，让App1仅在收到请求时执行逻辑，其余时间处于等待（休眠）状态：

方案1：使用Python内置XML-RPC（无额外依赖） #

XML-RPC是Python自带的远程过程调用框架，无需安装第三方包，适合快速实现简单的函数调用。

App1（常驻服务端）代码 #

from xmlrpc.server import SimpleXMLRPCServer

# 启动时加载大量数据与资源（示例）
loaded_data = {"key1": "value1", "large_dataset": [i for i in range(100000)]}

def process_request(param1, param2):
    # 使用加载的资源和传入参数计算结果
    result = f"基于资源{loaded_data['key1']}，处理参数{param1}和{param2}的结果"
    return result

if __name__ == "__main__":
    # 创建RPC服务器，绑定本地端口8000
    server = SimpleXMLRPCServer(("localhost", 8000))
    print("App1已启动，等待请求...")
    # 注册要被远程调用的函数
    server.register_function(process_request, "process_request")
    # 启动服务器，一直运行（无请求时处于阻塞等待状态，相当于休眠）
    server.serve_forever()

App2（临时客户端）代码 #

from xmlrpc.client import ServerProxy

if __name__ == "__main__":
    # 连接到App1的RPC服务器
    proxy = ServerProxy("http://localhost:8000/")
    # 传递参数调用远程函数
    response = proxy.process_request("参数A", "参数B")
    # 打印结果
    print("App1返回的结果：", response)

操作步骤 #

1. 先运行App1，它会加载资源并等待请求
2. 当需要调用时，运行App2，它会发送请求并接收结果

方案2：使用ZeroRPC（轻量高效，支持更多参数类型） #

ZeroRPC是一个轻量级的RPC框架，支持传递复杂数据类型，性能优于XML-RPC，需要先安装依赖：

pip install zerorpc

App1（常驻服务端）代码 #

import zerorpc

# 启动时加载资源
loaded_model = {"model_version": "v1.0", "weights": [0.1, 0.2, 0.3]}

class App1Service:
    def calculate(self, input_data):
        # 用加载的模型和传入参数计算
        result = sum(loaded_model["weights"]) + sum(input_data)
        return result

if __name__ == "__main__":
    server = zerorpc.Server(App1Service())
    server.bind("tcp://0.0.0.0:5555")
    print("App1已启动，等待请求...")
    server.run()  # 阻塞等待请求，无请求时处于休眠状态

App2（临时客户端）代码 #

import zerorpc

if __name__ == "__main__":
    client = zerorpc.Client()
    client.connect("tcp://localhost:5555")
    # 传递复杂参数（比如列表）
    result = client.calculate([1, 2, 3])
    print("App1返回的结果：", result)

方案3：基础Socket通信（自定义协议，无依赖） #

如果需要完全自定义通信逻辑，可以用Python原生Socket实现，需要自己处理数据序列化（比如用JSON）。

App1（常驻服务端）代码 #

import socket
import json

# 加载资源
loaded_config = {"threshold": 10, "allowed_values": ["a", "b", "c"]}

def handle_request(data):
    # 解析参数
    params = json.loads(data)
    param = params.get("param")
    # 处理逻辑
    if param in loaded_config["allowed_values"] and params.get("value") > loaded_config["threshold"]:
        return {"status": "success", "result": f"参数{param}符合要求，计算值为{params['value'] * 2}"}
    else:
        return {"status": "fail", "result": "参数不符合规则"}

if __name__ == "__main__":
    # 创建Socket对象
    server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server_socket.bind(("localhost", 9999))
    server_socket.listen(1)
    print("App1已启动，等待请求...")
    
    while True:
        # 等待客户端连接（阻塞状态，无请求时休眠）
        client_conn, addr = server_socket.accept()
        print(f"收到来自{addr}的请求")
        # 接收数据
        request_data = client_conn.recv(1024).decode("utf-8")
        # 处理请求
        response = handle_request(request_data)
        # 返回结果
        client_conn.send(json.dumps(response).encode("utf-8"))
        # 关闭连接
        client_conn.close()

App2（临时客户端）代码 #

import socket
import json

if __name__ == "__main__":
    client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    client_socket.connect(("localhost", 9999))
    # 构造请求参数
    request_params = {"param": "a", "value": 15}
    # 发送数据
    client_socket.send(json.dumps(request_params).encode("utf-8"))
    # 接收结果
    response_data = client_socket.recv(1024).decode("utf-8")
    response = json.loads(response_data)
    print("App1返回的结果：", response)
    # 关闭连接
    client_socket.close()

方案选择建议 #

• 优先选方案1：无依赖，代码简单，适合参数类型简单的场景
• 若需要传递复杂数据（如自定义对象、大数组），选方案2：ZeroRPC序列化能力更强，性能更好
• 若需要高度自定义通信协议，选方案3：完全可控，但需要自己处理数据解析和异常

内容的提问来源于stack exchange，提问作者FLOROID