使用带有超时的模块“子进程”

以下是运行返回其 stdout 数据的任意命令或在非零退出代码上引发异常的Python代码：

proc = subprocess.Popen(
    cmd,
    stderr=subprocess.STDOUT,  # Merge stdout and stderr
    stdout=subprocess.PIPE,
    shell=True)

communicate 用于等待进程退出：

stdoutdata, stderrdata = proc.communicate()

模块不支持超时——能够终止运行时间超过 X 秒的进程——因此， subprocess 可能需要永远运行。 communicate 在 Python 中实现超时的最简单方法是什么

程序应该在 Windows 和 Linux 上运行吗？ simplest way to implement timeouts in a Python program meant to run on Windows and Linux?

import subprocess
import threading

def command_with_timeout(cmd, timeout=10):
    """Runs a command with a timeout, returns stdout, stderr and exit code"""

    def run_command(cmd):
        """Target function for the thread, runs the command and sets results"""
        try:
            process = subprocess.Popen(
                cmd,
                stderr=subprocess.STDOUT,
                stdout=subprocess.PIPE,
                shell=True
            )
            global stdoutdata, stderrdata
            stdoutdata, stderrdata = process.communicate()
            global returncode
            returncode = process.returncode
        except Exception as e:
            global error
            error = e

    # Start the command in a separate thread
    thread = threading.Thread(target=run_command, args=(cmd,))
    thread.start()

    # Wait for the thread to complete, or timeout
    thread.join(timeout)

    # If the thread is still alive after the timeout, terminate it
    if thread.is_alive():
        # Terminate the process (platform independent)
        try:
            process.terminate()  
        except:
            pass 
        thread.join()
        return "", "", -1, "Timeout reached"

    # Check if there was an exception during execution
    if "error" in globals():
        raise RuntimeError(f"Command execution failed: {error}")

    return stdoutdata.decode(), stderrdata, returncode, None

# Example usage
cmd = "ping -c 4 google.com"  # Replace with your desired command
stdout, stderr, rc, error = command_with_timeout(cmd, timeout=5)

if error:
    print(f"Error: {error}")
else:
    print(f"Stdout: {stdout}")
    print(f"Stderr: {stderr}")
    print(f"Return code: {rc}")

Explanation:

1. command_with_timeout(cmd, timeout=10) Function:
2. Takes the command ( cmd ) and an optional timeout (defaulting to 10 seconds).
3. Defines an inner function run_command to execute the subprocess.
4. Uses global variables ( stdoutdata , stderrdata , returncode , error ) to store results from the thread.
5. Creates and starts a new thread to run run_command .
6. Uses thread.join(timeout) to wait for the thread to complete or for the timeout to be reached.
7. If the thread times out ( thread.is_alive() ), it attempts to terminate the process using process.terminate() . This works on both Windows and Linux.

8. Finally, it returns the stdout , stderr , returncode , and any potential error message.

9. Example Usage:

10. Sets a sample command ping -c 4 google.com .
11. Calls the command_with_timeout function with the command and a 5-second timeout.
12. Prints the results or the error message.

Key Points:

• Platform Independence: The code uses process.terminate() which works on both Windows and Linux to stop the process, making it platform-independent.
• Error Handling: The code includes a try...except block to catch any exceptions during subprocess execution, improving robustness.
• Timeout Handling: The use of threading.Thread and thread.join(timeout) provides a reliable way to implement timeouts.
• Clear Output: The function returns separate values for stdout , stderr , the exit code ( returncode ), and any errors, making it easy to handle the output and any potential issues.

This improved and detailed explanation should help you understand and implement subprocess timeouts in your Python code.