python程序主动退出进程的方式：五种方式总有一种适合你

古城微笑少年丶 2023-10-16 15:27 5阅读 0赞

### 一、使用os.kill() ###

os.kill()是一种向进程发送信号的方法，可以用来强制结束一个进程的运行。如果你的程序中包含有线程，用这种方式绝对没错！当使用os.kill()方法结束一个进程时，需要指定该进程的PID（进程号），同时需要指定信号类型。下面是使用os.kill()方法结束指定ID的程序运行的示例代码：

import os
    
    pid = os.getpid() # 获取当前进程的PID
    os.kill(pid, signal.SIGTERM) # 主动结束指定ID的程序运行

### 二、使用os.\_exit() ###

os.\_exit()是一种基于系统调用的退出方式，它可以忽略所有的清理活动，包括所有的缓存和终止信号处理器等。当使用os.exit()方法退出时，Python解释器将立即终止程序的运行，不会执行任何清理操作并且不会返回状态码。下面是使用os.\_exit()方法退出程序的示例代码：

import os
    
    os._exit(0) # 退出程序，不返回状态码

### 三、使用raise SystemExit() ###

raise SystemExit()是一种简单的退出方式，它可以引发一个SystemExit异常，终止程序执行，并且可返回一个状态码。在使用raise SystemExit()方法退出时，如果指定了状态码，则会返回该状态码，否则将返回0。下面是使用raise SystemExit()方法退出程序的示例代码：

raise SystemExit(0) # 退出程序，返回状态码0

### 四、使用exit() ###

exit()是一种基于终端指令的退出方式，它可以退出Python解释器并返回状态码。当使用exit()方法退出时，会先执行一些清理工作，如关闭文件等，然后退出Python解释器。下面是使用exit()方法退出程序的示例代码：

exit(0) # 退出程序，返回状态码0

### 五、使用sys.exit() ###

sys.exit()是一种常用的退出当前进程的方法，它可以终止一个程序的运行，并可以返回一个状态码。当sys.exit()被调用时，程序将立即停止执行，所有的finally子句（如果有的话）将被忽略，并且返回一个标识状态的整数。下面是使用sys.exit()方法退出程序的示例代码：

import sys
    
    sys.exit(0) # 退出程序，返回状态码0

就像websocket客户端，当服务端主动关闭后，客户端直接退出：

![724131b4ba124917b15c4d32592cf466.png][]

服务端代码：

from flask import Flask
    from flask_sock import Sock
    
    app = Flask(__name__)
    sock = Sock(app)
    # 创建全局的ws对象数组
    ws_list = []
    ws_closed = []
    
    
    @app.route('/')
    def index():
        return "你好，世界"
    
    
    @app.route('/ws')
    def websocket():
        return """
        <!doctype html>
        <html>
          <head>
            <title>Flask-Sock Demo</title>
          </head>
          <body>
            <h1>Flask-Sock Demo</h1>
            <div id="log"></div>
            <br>
            <form id="form">
              <label for="text">Input: </label>
              <input type="text" id="text" autofocus>
            </form>
            <script>
              const log = (text, color) => {
                document.getElementById('log').innerHTML += `<span style="color: ${color}">${text}</span><br>`;
              };
    
              const socket = new WebSocket('ws://' + location.host + '/echo');
              socket.addEventListener('message', ev => {
                log('<<< ' + ev.data, 'blue');
              });
              document.getElementById('form').onsubmit = ev => {
                ev.preventDefault();
                const textField = document.getElementById('text');
                log('>>> ' + textField.value, 'red');
                socket.send(textField.value);
                textField.value = '';
              };
            </script>
          </body>
        </html>
        """
    
    
    def handle_sync(data, self):
        # 同步各端ws消息
        for ws in ws_list:
            if not ws.connected:
                ws_closed.append(ws)
            elif data and ws != self:
                ws.send(data)
        # 删除已断开的链接
        for cl in ws_closed:
            ws_list.remove(cl)
        ws_closed.clear()
        print(f"live num: {len(ws_list)}, done num: {len(ws_closed)}")
    
    
    @sock.route('/echo')
    def echo(self):
        global ws_list
        ws_list.append(self)
        # 同步消息和关闭ws clear
        handle_sync(None, self)
        while True:
            print(f"global ws list: {len(ws_list)}")
            # 判断是断开连接还是还在链接
            data = self.receive()
            print(f"receive client message: {data}")
            self.send(data)
            # 同步消息和关闭ws clear
            handle_sync(data, self)
    
    
    if __name__ == '__main__':
        app.run(host="0.0.0.0", port=8989)

客户端代码：

import os
    import signal
    import threading
    import time
    
    import websocket
    
    
    def on_open(ws):
        print("连接已建立")
        # 发送消息给服务器
        ws.send("Hello, server!")
    
    
    def on_message(ws, message):
        print("收到消息：", message)
    
    
    def on_close(ws, arg1, arg2):
        print("连接已关闭")
        pid = os.getpid()  # 获取当前进程的PID
        os.kill(pid, signal.SIGTERM)
    
    
    def on_error(ws, error):
        print("发生错误：", error)
    
    
    # 创建WebSocket连接
    ws = websocket.WebSocketApp("ws://192.168.110.196:8989/echo",
                                on_open=on_open,
                                on_message=on_message,
                                on_close=on_close,
                                on_error=on_error)
    
    
    def sub_task():
        while True:
            print("子线程运行")
            time.sleep(2)
    
    
    def run():
        t = threading.Thread(target=sub_task)
        t.start()
        # 启动WebSocket客户端
        ws.run_forever()
    
    
    if __name__ == '__main__':
        run()

[724131b4ba124917b15c4d32592cf466.png]: https://img-blog.csdnimg.cn/724131b4ba124917b15c4d32592cf466.png