Python基礎之subprocess

前言

subprocess這個函數很好用，類似於控制台執行，功能很多，今天先介紹subprocess調用exe，並行調用兩個或兩個以上的exe。

Subprocess調用exe

調用exe有幾種方式，這里介紹一下subprocess。

p = subprocess.Popen(“./XXX.exe param1 param2”, stdin=subprocess.PIPE, stdout=subprocess.PIPE)

返回值為p.returncode，exe中打印出來的消息為output= p.communicate()[0]

那么有的程序運行時特別耗資源，容易卡死，所以設置一個超時時間，如果在規定時間內可以分析完畢，返回分析結果，如果超時了，殺死exe，返回默認結果。

def Func():
    p = subprocess.Popen("./XXX.exe", stdin=subprocess.PIPE, stdout=subprocess.PIPE)
    try:
        p.wait(timeout=SECONDS_TIMEOUT)
    except Exception as e:
        print("===== process timeout ======")
        p.kill()
        return None

    output= p.communicate()[0]
　　 err = p.communicate()[1]
　　 print(output)
　　 print(p.returncode)

 

 Subprocess並行調用兩個或兩個以上的exe

方法一：多線程

知識點：多線程，路徑分離，鎖定程序運行路徑

以並行調用三個exe為例。

1. readini.exe

功能：讀取同級目錄testini.ini里面的一個值，等待3秒，寫入同級目錄test1.ini里面

testini.ini

[TEST]
name = Sindy
age = 20
sex = girl
score = 90

 

 

2. rose.exe

功能：畫一朵玫瑰花，大概運行8S。

 

3. Usage.exe

功能：畫一個動態折線圖，時間20S。

 

CallEXEParaller.py

#!/usr/bin/env python
# _*_ coding: UTF-8 _*_
"""=================================================
@Project -> File    : 20200613 -> CallEXEParaller.py
@IDE     : PyCharm
@Author  : zihan
@Date    : 2020/6/13 14:20
@Desc    :目的：可以並行調用兩個或者多個EXE
================================================="""
import subprocess
import sys
import os
import threading


SECONDS_TIMEOUT = 1000


def execute_exe(exe_path, exe_param):
    folder_path, file_name = os.path.split(exe_path)
    # os.chdir(folder_path)
    p = subprocess.Popen(exe_path + " " + exe_param, stdin=subprocess.PIPE, stdout=subprocess.PIPE, cwd=folder_path)
    try:
        p.wait(timeout=SECONDS_TIMEOUT)
    except Exception as e:
        p.kill()
        # os.chdir(current_path)
        print("===== process timeout ======")
        return None
    except:
        s = sys.exc_info()
        str_error = "Error '%s' happened on line %d" % (s[1], s[2].tb_lineno)
        p.kill()
        # os.chdir(current_path)
        return None
    # os.chdir(current_path)
    output = p.communicate()[0]
    err = p.communicate()[1]
    print(output)
    print(p.returncode)


if __name__ == '__main__':
    current_path = os.getcwd()
    readini_abs_path = current_path + "\\" + r"EXEModule\ReadINI\readini.exe"
    rose_abs_path = current_path + "\\" + r"EXEModule\Rose\rose.exe"
    usage_abs_path = current_path + "\\" + r"EXEModule\CPUUsage\CPUUsage.exe"
    t1 = threading.Thread(target=execute_exe, args=(readini_abs_path, ""))
    t2 = threading.Thread(target=execute_exe, args=(rose_abs_path, ""))
    t3 = threading.Thread(target=execute_exe, args=(usage_abs_path, ""))
    t1.start()
    t2.start()
    t3.start()

 

注意subprocess的cwd參數，它可以指定當前exe運行在哪個路徑下。程序中執行的就是如果調用哪個exe，那么就在exe所在的目錄下執行程序。

 

方法二：多進程

在用多線程運行時發現了一個問題，如果在exe運行完后要作出相應的操作，那么必須得耗時最長的程序運行完后，在一起打印消息。這是因為p.communicate()[0]在阻塞的原因。解決此問題的方法就是用多線程進行調用。

CallEXEParallel.py

#!/usr/bin/env python
# _*_ coding: UTF-8 _*_
"""=================================================
@Project -> File    : 20200613 -> CallEXEParaller.py
@IDE     : PyCharm
@Author  : zihan
@Date    : 2020/6/13 14:20
@Desc    :目的：可以並行調用兩個或者多個EXE
        方法一：多進程方法調用，完美實現
        方法二：多線程調用，因為有p.communicate()[0]的存在，會導致打印阻塞，一直等到耗時長的程序執行完才會打印消息
================================================="""
import subprocess
import sys
import os
import threading
import multiprocessing


def execute_exe(exe_path, exe_param):
    print("aaa")
    folder_path, file_name = os.path.split(exe_path)
    try:
        p = subprocess.Popen(exe_path + " " + exe_param, stdin=subprocess.PIPE, stdout=subprocess.PIPE, cwd=folder_path)
    except:
        s = sys.exc_info()
        str_error = "Error '%s' happened on line %d" % (s[1], s[2].tb_lineno)
        print(str_error)
        return None
    output = p.communicate()[0]
    err = p.communicate()[1]
    print(output)
    print(p.returncode)
    if p.returncode == 0:
        print("pass")
    else:
        print("fail")


# 方法一：多進程方法調用
def multiprocess_test():
    current_path = os.getcwd()
    readini_abs_path = current_path + "\\" + r"EXEModule\ReadINI\readini.exe"
    rose_abs_path = current_path + "\\" + r"EXEModule\Rose\rose.exe"
    usage_abs_path = current_path + "\\" + r"EXEModule\CPUUsage\CPUUsage.exe"

    p1 = multiprocessing.Process(target=execute_exe, args=(readini_abs_path, ""))
    p1.start()
    p2 = multiprocessing.Process(target=execute_exe, args=(rose_abs_path, ""))
    p2.start()
    p3 = multiprocessing.Process(target=execute_exe, args=(usage_abs_path, ""))
    p3.start()

    p1.join()
    p2.join()
    p3.join()


# 方法二：多線程方法調用
def threading_test():
    current_path = os.getcwd()
    readini_abs_path = current_path + "\\" + r"EXEModule\ReadINI\readini.exe"
    rose_abs_path = current_path + "\\" + r"EXEModule\Rose\rose.exe"
    usage_abs_path = current_path + "\\" + r"EXEModule\CPUUsage\CPUUsage.exe"

    t1 = threading.Thread(target=execute_exe, args=(readini_abs_path, ""))
    t2 = threading.Thread(target=execute_exe, args=(rose_abs_path, ""))
    t3 = threading.Thread(target=execute_exe, args=(usage_abs_path, ""))

    t1.start()
    t2.start()
    t3.start()

    t1.join()
    t2.join()
    t3.join()


if __name__ == '__main__':
    # threading_test()
    multiprocess_test()

 

OK.