CSTOS

人生苦短 Python水深

0%

Python-并发性能

发表于 更新于 分类于
本文字数： 7.1k 阅读时长 ≈ 26 分钟

Python 性能提升、多线程相关的最佳实践。如：

  • gevent：通过 greenlet 实现协程，轻松进行并发同步或异步编程
  • threading：【标准库】基于线程的并行
  • multiprocessing：【标准库】基于进程的并行
  • asyncio：【标准库】异步 IO。new in python 3.4。
  • objgraph：查看内存中对象的数量，定位含有该对象的引用的所有代码的位置。
  • subprocess：调用 shell 命令的神器，创建一个子进程让其执行另外的程序，并与之通信

概述

  • 优先考虑数据结构与算法，合适的数据结构与算法能极大地提升性能。

基本概念

系统性能指标

并发数：系统能同时处理的请求/事务数量
响应时间：平均处理一个请求/事务所需的时长
QPS/TPS：每秒能处理请求/事务数量=并发数/响应时间

线程数

最佳线程数目 = （（线程等待时间 + 线程 CPU 时间）/线程 CPU 时间 ）* CPU 数目

  • CPU 密集型：操作内存处理的业务，一般线程数设置为：CPU 核数 + 1 或者 CPU 核数*2。核数为 4 的话，一般设置 5 或 8
  • IO 密集型：文件操作，网络操作，数据库操作，一般线程设置为：cpu 核数 / (1-0.9)，核数为 4 的话，一般设置 40

线程进程协程

进程：实际的运作单位，操作系统资源分配的最小单位，具有独立的内存空间，进程间数据不共享，创建的开销大。可以利用多核。

进程：一个运行的代码就是进程
程序：没有运行的代码

  • 如微信：整体是一个进程（但其实可能多进程），里面包含对各种资源的调用、内存管理、网络通讯等。
  • 多进程：Multiprocessing

线程：操作系统 CPU 调度执行的最小单位，依靠进程存在，是进程中的实际运作单位，一个进程至少一个线程，即主线程，多个线程共享内存（数据共享，共享全局变量），不能使用多核。

同一个进程中的多个线程之间可以并发执行，每个线程并行执行不同的任务

  • 单线程： Single Threaded
  • 多线程：Multi Threaded
  • 线程是一种对于非顺序依赖的多个任务进行解耦的技术。

协程：Coroutine 用户态的轻量级线程，调度由用户控制。

  • 具有自己的寄存器上下文和栈。协程调度切换时，将寄存器上下 文和栈保存到其他地方，在切回来的时候，恢复先前保存的寄存器上下文和栈，直接操作栈则基本没有内核切换的开销，可以不加锁的访问全局变量，所以上下文的切换非常快。

选择（不成文的准则）：

  • 多进程：CPU 计算密集的操作：如矩阵计算、解压缩、加密解密
  • 多线程：IO 密集的操作：IO 操作不占 CPU，如保存图像、文件处理、网络爬虫、数据库访问等。
    • 由于 GIL，对 CPU 密集的操作，由于计算工作多，多线程会触发 GIL 的竞争，消耗资源且没有太大速度提升，甚至可能由于切换任务过多导致减速。

多线程与多进程区别

  • 多线程可共享全局变量，共享内存空间，多进程不能，内存是独立的
  • 多线程中，所有子线程的进程中相同，多进程的不同
  • 同一进程的两线程间可直接交流；两进程通信需要一个中间代理
  • 创建新线程简单，创建新进程需要对父进程进行一次克隆

并发并行

并发：Concurrent：Python 由于 GIL 限制单进程只能并发，**基于单核 CPU 的多线程和异步 IO（多任务）同一时间内只能处理一件事件（但是它们有自己独特的机制来加快处理不同事件的能力），这个叫做并发。**包括多线程、异步 IO(多任务)、多进程

并行：Parallesim：Erlang, Golang 是天生支持并行的，只有调用多核 CPU 的多进程 (Multiprocessing) 是用来处理在物理上同时发生的任务的

  • 当你吃饭的时候突然有人给你打电话，如果此时你：
    1. 不接听电话，继续吃饭，等把饭吃完过后再来回电话，这个叫做同步。
    2. 接听电话后放下筷子停止进食，等通话完毕后再接着吃，这个叫做并发。（宏观上在一段时间内同时运行多个程序）
    3. 接听电话的同时继续进食，这个叫做并行。（在同一时刻能运行多个指令）
      1. 并行需要硬件支持，如多流水线、多核处理器或者分布式计算系统。

同步异步

同步：Synchronous：指事务完成的逻辑，顺序执行：先执行第一个，如果阻塞了，会一直等待，直到该事务完成才会执行下一个

异步：Asynchronous：不等待该事务的处理结果，直接处理第二个事务，能过状态、通知、回调来通知调用者处理结果。

gevent

gevent：通过 greenlet 实现协程，轻松进行并发同步或异步编程

  • 参考文档：官方文档
  • 基于协程的 Python 网络库，通过使用 greenlets 在 libev 上提供了一个高级的同步 API，将 Python 同步代码变成异步协程的库
  • 高并发：在同样的时间内执行更多有效的逻辑，减少无用的等待。
    • 以微线程 greenlet 为核心
    • epoll 事件监听机制
  • 每当 greenlet 遇到 IO 操作时，就自动切换到其他 greenlet，等 IO 操作完成，再在适当的时候切换回来继续执行。
  • monkey 机制：不修改原来的 Python 代码，在文件开头打一个 patch，它会自动替换原来的 thread、socket、time, multiprocess 等代码，全部变为 gevent 框架。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
import gevent


def f1():
    for i in range(5):
        print(f'run func:f1,index{i}')
        # 人为制造阻塞，gevent 运行到这里就会自动切换函数
        gevent.sleep(0)


def f2():
    for i in range(5):
        print(f'run func:f2,index:{i}')
        gevent.sleep(0)


t1 = gevent.spawn(f1)
t2 = gevent.spawn(f2)
gevent.joinall([t1, t2])

threading

threading：【标准库】基于线程的并行

  • 单进程下的多线程：
    • 默认启动的线程看作主线程，主线程可启动新的线程。
  • threading 模块具有线程锁、事件、条件变量、信号量等来控制线程对数据资源的操作，但使用不当还是可能产生问题：
    • 常用：将对资源的请求集中到一个线程，使用 queue 模块来向该线程应用进行请求，以队列形式进行线程间通信与协调，易设计、易读、可靠
1
2
# 返回当前线程的实例，主线程为 MainThread、若未用 LoopThread 命名子线程时，Python 会自动给线程命名为 Thread-1、Thread-2
threading.current_thread()
  • GIL
    • 由于 GIL 锁的存在，任何 Python 线程在执行前需先获得 GIL 锁，每执行 100 条字节码，解释器就自动释放 GIL 锁，让别的线程有机会执行，即给所有线程的执行代码都上了锁，那么多线程在 Python 中只能交替执行，100 个线程跑在 100 核 CPU 上，也只能用到 1 个核。
      • 而别的语言如 C、C++、Java，可以把全部核心跑满，即 N 核的 CPU，N 个死循环会把它占满。
    • GIL 是 Python 解释器设计的历史遗留问题，通常我们用的解释器是官方实现的 CPython，要真正利用多核，除非重写一个不带 GIL 的解释器。
    • SO，Python 中，可以使用多线程，不要指望用多线程有效利用多核，若一定要通过多线程利用多核，只能写 C 扩展 或 转多进程（多个 Python 进程各自有独立的 GIL 锁，互不影响）

基本方法

  • 启动一个线程就是把一个函数传入并创建 Thread 实例，然后调用 start() 开始执行
  • setDaemon(True)：将子线程变成了主线程的守护线程：当主线程结束后，守护线程会继续执行，只是主线程对其完成信息不再关心。
  • join：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
import multiprocessing
import threading
from threading import Lock,Thread
import time

def test(intervals=5):
    print(f"start {time.strftime('%Y-%m-%d %H-%M-%S',time.localtime())}")
    time.sleep(intervals)
    print(f"start {time.strftime('%Y-%m-%d %H-%M-%S',time.localtime())}")


if __name__=='__main__':
    # task=multiprocessing.Process(target=test,args=(10,))
    task=Thread(target=test,args=(10,))
    task.start()
    print(f"main start {time.strftime('%Y-%m-%d %H-%M-%S',time.localtime())}")
    time.sleep(3)
    print(f"main start {time.strftime('%Y-%m-%d %H-%M-%S',time.localtime())}")

调用

直接调用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
import threading
import time

def hello(name,time_sleep):
    print(f'Hello {name},start at {time.asctime()}')
    time.sleep(time_sleep)
    print(f'Hello {name},end at {time.asctime()}')


if __name__=='__main__':
    t1=threading.Thread(
      group=None, # 为了日后扩展ThreadGroup类实现而保留
      target=hello, # run方法调用的可调用对象
      args=('chen',3), # 参数元组
      daemon=None # 是否设置为守护线程（后台线程），守护线程是主线程退出时，会与主线程一同退出
      # 当创建的子线程是无限循环的话，就应该设置为守护线程。随着主线程的退出，子线程被强制退出，保证了整个python程序正常地，完整地退出。
      )
    t2=threading.Thread(target=hello,args=('zhang',5))

    t1.setName('tttt1')
    t1.start()
    t2.start()
    t2.join()   # join 等待 t2 先执行完
    print('test')
    print(t1.getName())

继承式调用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
import threading
import time
class MyThread(threading.Thread):
    def __init__(self,name):
        threading.Thread.__init__(self)
        self.name = name
    def run(self):
        print("Hello %s"%self.name)
        time.sleep(3)

if __name__ == "__main__":
    t1=MyThread("zhangsan")
    t2=MyThread("lisi")
    t1.start()
    t2.start()

线程池

  • 相比于 threading 等模块，该模块通过 submit 返回一个 future 对象
    • 主线程可获取某一个线程的状态、返回值
    • 当一个线程完成时，主线程能立即知道
    • 多线程与多进程的编码接口一致
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
# 模块，提供了ThreadPoolExecutor与ProcessPoolExecutor
from concurrent.futures import ThreadPoolExecutor
import threading
import time

# 定义一个准备作为线程任务的函数
def action(max):
    my_sum = 0
    for i in range(max):
        print(threading.current_thread().name + '  ' + str(i))
        my_sum += i
    return my_sum
# 创建一个包含2条线程的线程池
pool = ThreadPoolExecutor(max_workers=2)
with ThreaadPoolExecutor(max_workers=5) as t:
 task_1=t.submit(spider,1)
# 向线程池提交一个task, 50会作为action()函数的参数
task_1 = pool.submit(action, 50)
# 向线程池再提交一个task, 100会作为action()函数的参数
task_2 = pool.submit(action, 100)
# 判断future1代表的任务是否结束
print(future1.done())
# 判断future2代表的任务是否结束
print(task_2.done())
# 查看future1代表的任务返回的结果
print(task_1.result())
# 查看future2代表的任务返回的结果
print(future2.result())
# 关闭线程池
pool.shutdown()

wait(fs, # 需要执行的序列
 timeout=None, # 等待的最大时间，超过这个时间即使线程未执行完成也返回
 return_when=ALL_COMPLETED # wait返回结果的条件，默认为ALL_COMPLETED
 )

# as_completed()生成器，没有任务完成时会一直阻塞，除非设置了timeout，有任务完成时会yield这个任务，执行for循环下的语句后明治维新阻塞，循环到所有任务完成。
for task in as_completed(task_list):
 result=task.result()
 print(f'{result}')

# map
map(fn,
 *iterables,
 tiemout=None # 由于map是返回线程执行的结果，若timeout小于线程执行时间会抛出异常TimeoutError
   )
from concurrent.futures import ThreadPoolExecutor
import threading
import time
# 定义一个准备作为线程任务的函数
def action(max):
    my_sum = 0
    for i in range(max):
        print(threading.current_thread().name + '  ' + str(i))
        my_sum += i
    return my_sum
# 创建一个包含2条线程的线程池
with ThreadPoolExecutor(max_workers=2) as pool:
    # 向线程池提交一个task, 50会作为action()函数的参数
    future1 = pool.submit(action, 50)
    # 向线程池再提交一个task, 100会作为action()函数的参数
    future2 = pool.submit(action, 100)
    def get_result(future):
        print(future.result())
    # 为future1添加线程完成的回调函数
    future1.add_done_callback(get_result)
    # 为future2添加线程完成的回调函数
    future2.add_done_callback(get_result)
    print('--------------')

Lock

  • 多线程较为复杂，容易发生冲突，需要用锁来加以隔离，同时避免死锁的发生
  • 多个线程同时执行 lock.acquire() 时，只有一个线程能成功获取锁，然后继续执行代码，其他线程需等待直到获得锁为止。
  • 锁：
    • 好处：确保某段关键代码只由一个线路程从头到尾执行
    • 坏处：
      • 该段代码只能以单线程运行
      • 多个锁时，可能造成死锁，多个线程全部挂起，既不能执行，也无法结束，只能由操作系统强制终止。
1
2
threading.Lock()
try...finally 来确保一定会释放锁。

ThreadLocal

  • 多线程程序，尽量使用局部变量，全局变量的使用必须加锁
  • ThreadLocal 变量虽然是全局变量，但每个线程只能读写自己线程的独立副本，解决了参数在一个线程中各个函数之间互相传递的问题。
    • 即用一个参数名，在不同的线程中是不同的数据。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
import threading
# 创建全局 ThreadLocal 对象:
local_school = threading.local()
def process_student():
    # 获取当前线程关联的 student:
    std = local_school.student
    print('Hello, %s (in %s)' % (std, threading.current_thread().name))
def process_thread(name):
    # 绑定 ThreadLocal 的 student:
    local_school.student = name
    process_student()
t1 = threading.Thread(target= process_thread, args=('Alice',), name='Thread-A')
t2 = threading.Thread(target= process_thread, args=('Bob',), name='Thread-B')
t1.start()
t2.start()
t1.join()
t2.join()
1
2
Hello, Alice (in Thread-A)
Hello, Bob (in Thread-B)

获取线程结果

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
```python
class MyThread(threading.Thread):

    def __init__(self, func, args=()):
        super(MyThread, self).__init__()
        self.func = func
        self.args = args

    def run(self):
        self.result = self.func(*self.args)

    def get_result(self):
        try:
            return self.result   # 如果子线程不使用join方法，此处可能会报没有self.result的错误
        except Exception:
            return None

t = MyTread(函数名，args=(参数,))
t.start()
t.join()
t.get_result()     # 获取线程执行结果

队列

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
Queue.Queue(maxsize=0) #FIFO， 用来定义队列的长度，如果maxsize小于1就表示队列长度无限,
 Queue.LifoQueue(maxsize=0) #LIFO， 如果maxsize小于1就表示队列长度无限
 Queue.qsize() #返回队列的大小
 Queue.empty() #如果队列为空，返回True,反之False ，在线程间通信的过程中，可以通过此来给消费者等待信息
 Queue.full() # 如果队列满了，返回True,反之False，给生产者提醒
 Queue.get([block[, timeout]]) 读队列，timeout等待时间
 Queue.put(item, [block[, timeout]]) 写队列，timeout等待时间
 Queue.queue.clear() 清空队列

task_done()#意味着之前入队的一个任务已经完成。由队列的消费者线程调用。每一个get()调用得到一个任务，接下来
的task_done()调用告诉队列该任务已经处理完毕如果当前一个join()正在阻塞，它将在队列中的所有任务都处理完时恢复
执行（即每一个由put()调用入队的任务都有一个对应的task_done()调用）。

join()#阻塞调用线程，直到队列中的所有任务被处理掉。只要有数据被加入队列，未完成的任务数就会增加。当消费者
线程调用task_done(（意味着有消费者取得任务并完成任务），未完成的任务数就会减少。当未完成的任务数降到0，
join()解除阻塞。

multiprocessing

multiprocessing：【标准库】基于进程的并行

  • 关于多进程的进程数量，推荐 multiprocessing.cpu_count() * 2 + 1
  • 多线程与多进程
    • GIL 一定程度地限制了 Threading 多线程，而目前电脑大多数是多核处理器，Multiprocessing 多进程能让电脑更有效率地分配任务给每一个处理器。解决了多线程的弊端，并有效提高效率。
    • Process：可分布到多台机器上，相比 Thread 更稳定
    • Thread：最多分布到同一台机器，多个 CPU 上
  • managers 支持将多进程分布到多台机器上，一个服务进程可以作为调度者，将任务分布到其他多个进程中，依靠网络通信。
    • 实现了进程间数据的共享，即可多个进程修改同一份数据
    • 封装良好，避免了网络通信细节，容易使用
    • 实现分布式计算
  • multiprocessing.shared_memory：可从进程直接访问的共享内存
  • 进程间数据传递：Queue、Pipes
    • Python 的 multiprocessing 模块包装了底层的机制，提供了 Queue、Pipes 等多种方式来交换数据。
    • 在 Unix/Linux 下，multiprocessing 模块封装了 fork() 调用，使我们不需要关注 fork() 的细节。由于 Windows 没有 fork 调用，因此，multiprocessing 需要“模拟”出 fork 的效果，父进程所有 Python 对象都必须通过 pickle 序列化再传到子进程去，所有，如果 multiprocessing 在 Windows 下调用失败了，要先考虑是不是 pickle 失败了。
  • Lock 进程锁，主要用在输出到屏幕的时候独占屏幕，即多份数据不会出现打印一半就去打印别的数据

基本使用

1
2
3
4
5
6
7
8
9
10
11
12
import multiprocessing as mp
threads=[]
for i in path:
    threads.append(mp.Process(target=md5sum,args=(i,)))
#创建一个 multiprocessing.process.Process 对象

#执行
for m in threads:
    m.start()
#回收
for m in threads:
  m.join()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
import multiprocessing
import threading
import time

def hello(name,time_sleep):
    print(f'Hello {name},start at {time.asctime()}')
    time.sleep(time_sleep)
    print(f'Hello {name},end at {time.asctime()}')

def create_thread():
    t1=threading.Thread(target=hello,args=('chen',3))
    t2=threading.Thread(target=hello,args=('zhang',5))

    t1.setName('tttt1')
    t1.start()
    t2.start()
    t2.join()   # join 等待 t2 先执行完
    print('test')
    print(t1.getName())
    return 0
if __name__=='__main__':
    print(create_thread())

Process

1
2
3
4
5
6
7
8
9
10
11
from multiprocessing import Process  # 进程类
import os
# 子进程要执行的代码
def run_proc(name):
    print('Run child process %s (%s)...' % (name, os.getpid()))
if __name__=='__main__':
    print('Parent process %s.' % os.getpid())
    p = Process(target=run_proc, args=('test',))  # 子进程，执行函数与函数的参数
    p.start()   # 启动子进程
    p.join()  # 等子进程结束后再继续往下运行，用于进程间的同步
    print('Child process end.')

Pool

  • 若要启动大量子进程，常采用进程池的方式
    对 Pool 对象调用 join() 方法会等待所有子进程执行完毕，调用 join() 之前必须先调用 close()，调用 close() 之后就不能继续添加新的 Process 了。
  • 默认大小为 4，4 个进程，p = Pool(5) 时，就可跑 5 个进程。

Manager

  • 用于获取子进程的结果：多进程共享全局变量
  • 支持类型：list, dict, Namespace, Lock, RLock, Semaphore, BoundedSemaphore, Condition, Event, Queue, Value 和 Array
  • 进程通信：(进程之间传递数据) 用进程队列 (multiprocessing.Queue()，单向通信)，管道 (multiprocessing.Pipe() ，双向通信)。
1
如果要共享全局变量需要(multiprocessing.Value("d",10.0)，数值)(multiprocessing.Array("i",[1,2,3,4,5])，数组)(multiprocessing.Manager().dict()，字典)(multiprocessing.Manager().list(range(5)))。

Queue

  • 多进

celery

celery：异步库

asyncio

asyncio：【标准库】异步 IO。new in python 3.4。

  • 由于 GIL，Python 在多核操作饱受诟病，而对于 IO 密集型，python 的异步处理提升成百上千倍的效率，弥补了 python 性能方面的短板。如最新的微服务框架 japronto，resquests per second 可达百万级。
  • event_loop：事件循环，将需要执行的协程放到 EventLoop 中执行
  • coroutine 协程：这里指使用 async 定义的函数，调用时不会立即执行函数，而是返回协程对象给 event_loop 调用。
  • task 任务：对协程的封装，包含了任务的各种状态。
  • future：代表将来执行或没有执行的任务的结果，与 task 一样
    • Pending
    • Running
    • Done
    • Cacelled
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
import asyncio
import time

# 声明协程函数
async def add(x,y):
    print(f'add {x},{y}')
    await asyncio.sleep(2)
    # time.sleep(2)
    return x+y

async def add2(x,y):
    print(f'add {x},{y}')
    await asyncio.sleep(5)
    # time.sleep(5)
    return x+y
now = time.time()
loop=asyncio.get_event_loop()
tasks=[
    loop.create_task(add(2,3)),
    loop.create_task(add2(3,4)),
    loop.create_task(add(4,2)),
]
loop.run_until_complete(asyncio.wait(tasks))
for task in tasks:
    print(task.result())
# loop.run_until_complete(add())  # 将函数注册到 loop
# loop.run_until_complete(asyncio.wait(tasks))  # 将函数注册到 loop
print('cost:',time.time()-now)

aiohttp

aiohttp：并发爬虫，可以看作异步的 requests

objgraph

objgraph：查看内存中对象的数量，定位含有该对象的引用的所有代码的位置。

  • 查找内存泄露：由于 Python 的引用计数机制，如果程序中不再被使用的对象的引用一直被占有，就会产生内存泄漏。
1
2
3
import objgraph
objgraph.show_most_common_types() # 程序中前 20 个最普遍的对象
objgraph.show_growth()

subprocess

subprocess：允许你生成新的进程，连接它们的输入、输出、错误管道，并且获取它们的返回三。

  • 调用 shell 命令的神器，创建一个子进程让其执行另外的程序，并与之通信
  • 避免使用os.system，因为它容易受到命令注入攻击
  • 不要再用 os.system、os.spawn 了，subprocess 就是为打造一个统一模块来进行进程创建的
    • 提供进程创建相关函数的所有功能
    • 跨进程异常优化：子进程的异常会在父进程中再次抛出，以便检测子进程执行情况
    • 提供用于在 fork 和 exec 之间执行自定义代码的钩子
    • 没有隐式调用/bin/sh，这意味着不需要对危险的 shell meta characters 进行转义；
    • 支持文件描述符重定向的所有组合；
    • 使用 subprocess 模块，可以控制在执行新程序之前是否应关闭所有打开的文件描述符；
    • 支持连接多个子进程 ;
      • 支持 universal newline；
    • 支持 communication() 方法，它使发送 stdin 数据以及读取 stdout 和 stderr 数据变得容易，而没有死锁的风险;
  • run 与 Popen
    • run 函数在启动子进程后，该函数会一直阻塞，直到命令行执行完毕
    • Popen 启动子进程之后，父进程并不需要阻塞等待子进程结束，而是可以去做别的任务
  • Popen 类
1
2
3
4
5
6
7
8
9
10
11
import subprocess

# 父进程等待子进程完成
subprocess.call()
subprocess.check_all()

# 父进程等待子进程完成，返回子进程向标准输出的输出结果
subprocess.check_output()

result = subprocess.run(['echo', 'Hello World'], capture_output=True, text=True)  
print(result.stdout)

Popen：执行命令并返回进程对象
call：执行命令
check_output：获取命令输出

Popen

  • Popen 对象创建后，主程序不会自动等待子进程完成。我们必须调用对象的 wait() 方法，父进程才会等待 (也就是阻塞 block)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
class subprocess.Popen(args, 
                       bufsize=-1, 
                       executable=None, 
                       stdin=None, 
                       stdout=None, 
                       stderr=None, 
                       preexec_fn=None, 
                       close_fds=True, 
                       shell=False, 
                       cwd=None, 
                       env=None, 
                       universal_newlines=False, 
                       startupinfo=None, 
                       creationflags=0,
                       restore_signals=True, 
                       start_new_session=False, 
                       pass_fds=(),
                       *, 
                       encoding=None, 
                       errors=None)


class Popen(args, bufsize=0, executable=None, stdin=None, stdout=None, stderr=None, preexec_fn=None, close_fds=False, shell=False, cwd=None, env=None, universal_newlines=False, startupinfo=None, creationflags=0)

def subprocess.run(args, # string 或 sequence（优先，可处理任何必需的参数转义和引用），如果传递的是字符串，则shell需要为Fasle，否则会直接看做要执行的程序名称，而不能指定任何参数
                   *,
                   stdin=None,
                   input=None, # ，该参数传递给Popen.communicate()，然后传递给子进程的stdin。该参数数据类型应为字节序列（bytes）;但如果指定了encoding , errors参数或则 text=True，参数则必须为字符串。使用该参数时，内部Popen对象，将使用stdin = PIPE自动创建该对象，不能同时使用stdin参数。
                   stdout=None,
                   stderr=None,
                   capture_output=False, # True则将捕获stdout和stderr，自动调用=PIPE创建标准输出和标准错误对象。
                   # 传递`stdout`和`stderr`参数时不能同时传递`capture_output`参数。如果希望捕获并将两个stream合并为一个，使用`stdout=PIPE`和`stderr = STDOUT`。
                   shell=False, # 要注意可能潜在的安全问题，需要确保所有空格和元字符都被适当地引用，以避免shell注入漏洞
                   cwd=None,
                   timeout=None,  # 该参数传递给Popen.communicate()，如果指定时间之后子进程仍未结束，子进程将被kill，并抛出TimeoutExpired异常。
                   check=False, # 如果`check=True`，并且进程以非零退出代码退出，则将抛出`CalledProcessError`异常
                   encoding=None,
                   errors=None,
                   text=None,
                   env=None,  # 通过传递mappings对象，给子进程提供环境变量，该参数直接传递给Popen函数。
                   universal_newlines=None,
                   *other_popen_kwargs)->subprocess.CompletedProcess:
  pass


如果设置 stderr=subprocess.STDOUT ，标准错误将被合并到标准输出中。
result = subprocess.run(['python', 'my.py'], stderr=subprocess.STDOUT, stdout=subprocess.PIPE)
  • 方法
  • args, 调用该进程的参数，同 subprocess.run(args，***) 中的 args；
    returncode，当值为 0 时，代表子进程执行成功；负值 -N 指示进程被 signal N 所终止 (POSIX only); None 代表未终止；
    stdout,stderr ,代表子进程的标准输出和标准错误；
    check_returncode(), check 子进程是否执行成功，若执行失败将抛出异常；
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
import subprocess

def run_process(command):
    process = subprocess.Popen(command, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    output, error = process.communicate()
    return output

commands = ['echo "Hello"', 'echo "World"']
processes = []

for command in commands:
    process = subprocess.Popen(command, shell=True)
    processes.append(process)

for process in processes:
    process.wait()



# Importing subprocess 
import subprocess

# Your command 
cmd = "python other_script.py"

# Starting process
process = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)

# Getting the output and errors of the program
stdout, stderr = process.communicate()

# Printing the output
print(stdout)

process = subprocess.Popen(['your_program', 'arg1', 'arg2'])

# 等待子进程完成
process.wait()

# 检查返回代码
if process.returncode == 0:
    print("程序成功执行！")
else:
    print("程序执行失败，返回代码:", process.returncode)

import subprocess

process = subprocess.Popen(['python', 'interactive_script.py'], stdin=subprocess.PIPE, stdout=subprocess.PIPE)

# 向子程序写入数据
process.stdin.write(b'Hello\n')
process.stdin.flush()

# 获取输出
output = process.stdout.readline()
print('输出:', output.decode())

属性

  • args：shell 命令，可以是字符串或者序列类型（如：list，元组）

  • bufsize：缓冲区大小。当创建标准流的管道对象时使用，默认 -1。
    0：不使用缓冲区
    1：表示行缓冲，仅当 universal_newlines=True 时可用，也就是文本模式
    正数：表示缓冲区大小
    负数：表示使用系统默认的缓冲区大小。

  • stdin, stdout, stderr：分别表示程序的标准输入、输出、错误句柄

  • preexec_fn：只在 Unix 平台下有效，用于指定一个可执行对象（callable object），它将在子进程运行之前被调用

  • shell：如果该参数为 True，将通过操作系统的 shell 执行指定的命令。

  • cwd：用于设置子进程的当前目录。

  • env：用于指定子进程的环境变量。如果 env = None，子进程的环境变量将从父进程中继承。

方法

  • poll(): 检查进程是否终止，如果终止返回 returncode，否则返回 None。
  • wait(timeout): 等待子进程终止。
  • communicate(input,timeout): 和子进程交互，发送和读取数据。
  • send_signal(singnal): 发送信号到子进程 。
  • terminate(): 停止子进程,也就是发送 SIGTERM 信号到子进程。
  • kill(): 杀死子进程。发送 SIGKILL 信号到子进程。相当于 kill -9 pid
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
import time
import subprocess

proc = subprocess.Popen(['sleep', '3'])

while proc.poll() is None:
   print('Do something others...')
   time.sleep(1)

print('Return Code: ', proc.poll())


from subprocess import Popen, TimeoutExpired

proc = Popen(['sleep', '10'])

try:
    proc.communicate(timeout=1)
except TimeoutExpired:
    proc.terminate()
    proc.wait()

print('子进程退出状态：', proc.poll())

Popen

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
import subprocess

# 执行命令
process = subprocess.Popen(["ls", "-l"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)

# 读取标准输出和错误
out, err = process.communicate()

print("标准输出:")
print(out)

print("标准错误:")
print(err)


import subprocess

s = subprocess.Popen("ipconfig", stdout=subprocess.PIPE, shell=True)
print(s.stdout.read().decode("GBK"))


import subprocess

s = subprocess.Popen("python", stdout=subprocess.PIPE, stdin=subprocess.PIPE, shell=True)
s.stdin.write(b"import os\n")
s.stdin.write(b"print(os.environ)")
s.stdin.close()

out = s.stdout.read().decode("GBK")
s.stdout.close()
print(out)


import subprocess

# cwd指定执行外部命令的工作目录
result = subprocess.run(["ls", "-l"], stdout=subprocess.PIPE, text=True, cwd="/path/to/directory")
print(result.stdout)


import subprocess

filename = "example.txt"
result = subprocess.run(["cat", filename], stdout=subprocess.PIPE, text=True)
print(result.stdout)


import subprocess

# 标准输入
input_data = "Hello, subprocess!"
result = subprocess.run(["grep", "subprocess"], input=input_data, stdout=subprocess.PIPE, text=True)
print(result.stdout)


import subprocess

# 标准输出
output_file = open("output.txt", "w")
result = subprocess.run(["ls", "-l"], stdout=output_file, text=True)
output_file.close()

import subprocess

# 标准错误
result = subprocess.run(["ls", "/nonexistent"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
print("标准输出:")
print(result.stdout)
print("标准错误:")
print(result.stderr)

import subprocess

# 返回码：0表示命令成功执行，非零则有错误
result = subprocess.run(["ls", "/nonexistent"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
if result.returncode != 0:
    print("命令执行失败。")
    print("标准错误:")
    print(result.stderr)


import subprocess

# 错误输出
result = subprocess.run(["ls", "/nonexistent"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
if result.returncode != 0:
    print("命令执行失败。")
    print("错误信息:")
    print(result.stderr)


import subprocess

# 创建第一个命令的进程
process1 = subprocess.Popen(["ls", "/path/to/directory"], stdout=subprocess.PIPE, text=True)

# 创建第二个命令的进程，将第一个命令的输出连接到它的输入
process2 = subprocess.Popen(["grep", "search_term"], stdin=process1.stdout, stdout=subprocess.PIPE, text=True)

# 从第二个命令的标准输出中读取结果
result = process2.communicate()[0]
print(result)


import subprocess

# 创建命令进程
process = subprocess.Popen(["python", "-u"], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, universal_newlines=True)

# 写入数据到标准输入
process.stdin.write("print('Hello from child process')\n")
process.stdin.flush()

# 读取并打印标准输出
output, errors = process.communicate()
print("标准输出:")
print(output)

# 打印标准错误
print("标准错误:")
print(errors)


import subprocess

# 超时处理
try:
    result = subprocess.run(["sleep", "10"], timeout=5, stdout=subprocess.PIPE, text=True)
    print(result.stdout)
except subprocess.TimeoutExpired:
    print("命令执行超时。")


import subprocess

# 使用Shell执行命令
result = subprocess.run("ls -l | grep .txt", shell=True, stdout=subprocess.PIPE, text=True)
print(result.stdout)


import time
import subprocess

def cmd(command):
    subp = subprocess.Popen(command,shell=True,stdout=subprocess.PIPE,stderr=subprocess.PIPE,encoding="utf-8")
    subp.wait(2)
    if subp.poll() == 0:
        print(subp.communicate()[1])
    else:
        print("失败")



cmd("java -version")
cmd("exit 1")

异常处理

  • SubprocessError：基类
    • TimeoutExpired：等待子进程超时时抛出
    • CalledProcessError：check_call() 或 check_output() 返回非零状态值时抛出
  • 属性：
    • cmd , 该子进程的命令
    • output , 子进程所 capture 的标准输出 (如调用 run() 或则 check_output())，否则为 None
    • stdout, output 的别名
    • stderr, 子进程所 capture 的标准错误 (如调用 run()) ,否则为 None
    • TimeoutExpired 还包括 timeout，指示所设置的 timeout 的值；CalledProcessError 则还包括属性 returncode;
相关文章
给博主来一杯卡布奇诺