从Linux命令行与运行中Python脚本通信的最佳实践及库推荐
没问题,这个需求太常见了——之前直接杀进程确实简单,但要优雅地先做清理操作(比如你说的pickle保存),甚至未来要发更多指令,就得用更合适的通信方式。我来给你梳理几个实用的方案,附带上代码示例和相关库:
1. 信号处理(最适合简单的「终止前清理」场景)
Linux的信号机制是最轻量化的方式,比如SIGTERM信号不是强制杀死进程,而是给进程发一个“该退出了”的通知,你的Python脚本可以捕获这个信号,先执行pickle操作再退出。
Python里用内置的signal模块就能实现,举个例子:
import signal import pickle import time import sys # 定义信号处理函数 def handle_terminate(signum, frame): print("收到终止信号,开始执行pickle操作...") # 这里替换成你要保存的特定内容 data_to_save = {"status": "finished", "timestamp": time.time()} with open("saved_data.pkl", "wb") as f: pickle.dump(data_to_save, f) print("保存完成,退出脚本") sys.exit(0) # 注册SIGTERM信号的处理函数 signal.signal(signal.SIGTERM, handle_terminate) # 模拟脚本的主运行逻辑 try: while True: print("脚本正在运行...") time.sleep(2) except KeyboardInterrupt: # 处理Ctrl+C的SIGINT信号,也可以在这里加清理逻辑 handle_terminate(signal.SIGINT, None)
然后在命令行里,你不用kill -9(强制杀死),而是发送SIGTERM信号:
kill -TERM <你的Python进程PID>
这种方式优点是简单、开销小,完全满足你当前的终止前pickle需求。
2. IPC通信(适合未来的复杂指令场景)
如果以后你需要给脚本发更多类型的指令(比如暂停、修改运行参数、获取当前状态),信号就不够用了,这时候可以用进程间通信(IPC)的方式,常用的有两种:
2.1 命名管道(FIFO)
命名管道是文件系统里的一个特殊文件,命令行可以往里面写指令,你的Python脚本持续监听这个管道的内容。
步骤:
- 先在命令行创建命名管道:
mkfifo my_script_pipe
- Python脚本监听管道:
import pickle import time import os PIPE_PATH = "my_script_pipe" def main(): # 打开管道,持续读取 with open(PIPE_PATH, "r") as pipe: while True: command = pipe.readline().strip() if not command: time.sleep(0.5) continue print(f"收到命令: {command}") if command == "terminate": print("执行pickle保存...") data_to_save = {"status": "terminated", "time": time.time()} with open("saved_data.pkl", "wb") as f: pickle.dump(data_to_save, f) print("保存完成,退出") os.unlink(PIPE_PATH) # 删除管道 break # 可以在这里添加其他命令的处理逻辑 elif command == "status": print("脚本当前运行正常") if __name__ == "__main__": main()
- 命令行发送指令:
echo "terminate" > my_script_pipe # 或者查询状态 echo "status" > my_script_pipe
2.2 Unix套接字
Unix套接字比命名管道更灵活,支持双向通信,适合更复杂的交互。Python的内置socket模块就能实现:
脚本端代码:
import socket import pickle import time import os SOCKET_PATH = "/tmp/my_script_socket" # 先清理可能存在的旧套接字 if os.path.exists(SOCKET_PATH): os.unlink(SOCKET_PATH) def main(): # 创建Unix域套接字 sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) sock.bind(SOCKET_PATH) sock.listen(1) print("等待命令连接...") conn, addr = sock.accept() with conn: while True: data = conn.recv(1024).decode().strip() if not data: break print(f"收到命令: {data}") if data == "terminate": print("执行pickle保存...") data_to_save = {"status": "terminated", "time": time.time()} with open("saved_data.pkl", "wb") as f: pickle.dump(data_to_save, f) conn.sendall(b"保存完成,即将退出") break elif data == "status": conn.sendall(b"脚本运行中") os.unlink(SOCKET_PATH) if __name__ == "__main__": main()
命令行可以用nc工具发送指令(如果没装可以用apt install netcat-openbsd):
# 发送终止命令 echo "terminate" | nc -U /tmp/my_script_socket # 查询状态 echo "status" | nc -U /tmp/my_script_socket
3. 现成的库(简化复杂场景开发)
如果你的需求越来越复杂,比如要做守护进程、多脚本协作,这些现成的库能帮你省不少事:
- python-daemon:专门用来把Python脚本做成守护进程,内置了信号处理的封装,同时支持标准的守护进程IPC机制
- pyzmq:基于ZeroMQ的消息库,跨平台支持各种IPC模式,命令行可以用
zmq工具发送消息,适合多进程、分布式的场景 - celery:如果你的脚本是任务型的,Celery可以让你通过命令行发送任务控制指令(比如终止任务、查看任务状态),不过这个库相对重一些,适合大型项目
总结
- 只是终止前做pickle:优先用信号处理,最简单高效
- 未来要发多种指令:选命名管道(轻量)或Unix套接字(灵活)
- 复杂场景:直接用现成的库减少开发量
内容的提问来源于stack exchange,提问作者jsstuball




