耿雨飞的个人博客

[[ slide_text ]]

[[ item.c ]]

Python 多进程、多线程、协程

发布时间：2023-03-22 作者：gengyufei 来源：gengyufei个人博客

Python 并发编程

# 概述

教科书上最经典的一句话是“**进程是资源分配的最小单位，线程是CPU调度的最小单位**”。

**进程**：**一个运行的程序（代码）就是一个进程**，没有运行的代码叫程序，**进程是系统资源分配的最小单位**，**进程拥有自己独立的内存空间，所以进程间数据不共享，开销大**。

**线程**：**CPU调度执行的最小单位，也叫执行路径，不能独立存在，依赖进程存在,****一个进程至少有一个线程，叫主线程，而多个线程共享内存 (数据共享，共享全局变量)，从而极大地提高了程序的运行效率**。

**协程**：**是一种用户态的轻量级线程，协程的调度完全由用户控制**。**协程拥有自己的寄存器上下文和栈**。**协程调度切换时，将寄存器上下 文和栈保存到其他地方，在切回来的时候，恢复先前保存的寄存器上下文和栈，直接操作栈则基本没有内核切换的开销，可以不加锁的访问全局变量，所以上下文的切换非常快**。

![](https://tcs-devops.aliyuncs.com/storage/112q9746c792285705f17677c5a0626ad56f?Signature=eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJBcHBJRCI6IjVlNzQ4MmQ2MjE1MjJiZDVjN2Y5YjMzNSIsIl9hcHBJZCI6IjVlNzQ4MmQ2MjE1MjJiZDVjN2Y5YjMzNSIsIl9vcmdhbml6YXRpb25JZCI6IiIsImV4cCI6MTY4MDA1MzI0NSwiaWF0IjoxNjc5NDQ4NDQ1LCJyZXNvdXJjZSI6Ii9zdG9yYWdlLzExMnE5NzQ2Yzc5MjI4NTcwNWYxNzY3N2M1YTA2MjZhZDU2ZiJ9.2i_UTeNzYrP4lH0rTIFLPd5CP9fD-eI8kIFsN931nJ0)

# 进程、线程、协程的关系
## 进程与线程的关系
线程与进程的区别可以归纳为以下4点：
1. **地址空间和其它资源**（如打开文件）：**进程间相互独立**，**同一进程的各线程间共享**。某进程内的线程在其它进程不可见。
2. **通信**：
 - 进程间通信IPC
 - 线程间可以直接读写进程数据段（如全局变量）来进行通信——需要进程同步和互斥手段的辅助，以保证数据的一致性。
3. **调度和切换**：**线程上下文切换比进程上下文切换要快得多**。
4. 在多线程OS中，进程不是一个可执行的实体。

## 多进程与多线程的比较
**进程**和**线程**还可以**类比为火车和车厢**：
- **线程在进程下行进**（单纯的车厢无法运行）
- **一个进程可以包含多个线程**（一辆火车可以有多个车厢）
- **不同进程间数据很难共享**（一辆火车上的乘客很难换到另外一辆火车，比如站点换乘）
- **同一进程下不同线程间数据很易共享**（A车厢换到B车厢很容易）
- **进程要比线程消耗更多的计算机资源**（采用多列火车相比多个车厢更耗资源）
- **进程间不会相互影响**，但是**一个线程挂掉将导致整个进程挂掉**（一列火车不会影响到另外一列火车，但是如果一列火车上中间的一节车厢着火了，将影响到该趟火车的所有车厢）
- **进程可以拓展到多机，线程最多适合多核**（不同火车可以开在多个轨道上，同一火车的车厢不能行进的不同的轨道上）
- **进程使用的内存地址可以上锁，即一个线程使用某些共享内存时，其他线程必须等它结束，才能使用这一块内存**。（比如火车上的洗手间）－”互斥锁（mutex）”
- **进程使用的内存地址可以限定使用量**（比如火车上的餐厅，最多只允许多少人进入，如果满了需要在门口等，等有人出来了才能进去）－“信号量（semaphore）”

## 协程与线程
1. 一个线程可以拥有多个协程，一个进程也可以单独拥有多个协程，这样python中则能使用多核CPU。

2. 线程进程都是同步机制，而协程则是异步。

3. 协程能保留上一次调用时的状态，每次过程重入时，就相当于进入上一次调用的状态。

## 线程、协程在python中的使用
- 多线程一般是使用threading库，完成一些IO密集型并发操作。多线程的优势是切换快，资源消耗低，但一个线程挂掉则会影响到所有线程，所以不够稳定。现实中使用线程池的场景会比较多
- 协程一般是使用gevent库。协程在tornado的运用较多，使用协程让tornado做到单线程异步。所以协程使用的地方最多的是在web应用上。

# 多进程模块 Multiprocessing
## 概述
**Python的threading包主要运用多线程的开发，但由于GIL的存在，Python中的多线程其实并不是真正的多线程**，如果**想要充分地使用多核CPU的资源**，大部分情况**需要使用多进程**。在Python 2.6版本的时候引入了multiprocessing包，它完整的复制了一套threading所提供的接口方便迁移。唯一的不同就是它使用了多进程而不是多线程。**每个进程有自己的独立的GIL，因此也不会出现进程之间的GIL争抢**。
借助这个multiprocessing，你可以轻松完成从单进程到并发执行的转换。multiprocessing支持子进程、通信和共享数据、执行不同形式的同步，提供了Process、Queue、Pipe、Lock等组件。

> 除了应对Python的GIL以外，产生multiprocessing的另外一个原因时Windows操作系统与Linux/Unix系统的不一致。
Unix/Linux操作系统提供了一个fork()系统调用，它非常特殊。普通的函数，调用一次，返回一次，但是fork()调用一次，返回两次，因为操作系统自动把当前进程（父进程）复制了一份（子进程），然后，分别在父进程和子进程内返回。子进程永远返回0，而父进程返回子进程的ID。这样做的理由是，一个父进程可以fork出很多子进程，所以，父进程要记下每个子进程的ID，而子进程只需要调用getpid()就可以拿到父进程的ID。

## 示例
```
import os
import threading
import multiprocessing

# Main
print('Main:', os.getpid())

# worker function
def worker(sign, lock):
    lock.acquire()
    print(sign, os.getpid())
    lock.release()

# Multi-thread
record = []
lock = threading.Lock()

# Multi-process
record = []
lock = multiprocessing.Lock()

if __name__ == '__main__':
    for i in range(5):
        thread = threading.Thread(target=worker, args=('thread', lock))
        thread.start()
        record.append(thread)

for thread in record:
        thread.join()
    
    for i in range(5):
        process = multiprocessing.Process(target=worker, args=('process', lock))
        process.start()
        record.append(process)
    
    for process in record:
        process.join()

# 结果打印
Main: 521124
thread 521124
thread 521124
thread 521124
thread 521124
thread 521124
process 521130
process 521131
process 521134
process 521132
process 521133
```

## pipe 管道
**Pipe可以是单向(half-duplex)，也可以是双向(duplex)**。我们**通过```mutiprocessing.Pipe(duplex=False)```创建单向管道 (默认为双向)**。

**一个进程从PIPE一端输入对象，然后被PIPE另一端的进程接收**，单向管道只允许管道一端的进程输入，而双向管道则允许从两端输入。

下面的程序展示了Pipe的使用（这里的Pipe是双向的）:
```
import multiprocessing as mul

def proc1(pipe):
    pipe.send('hello')
    print('proc1 rec:', pipe.recv())

def proc2(pipe):
    print('proc2 rec:', pipe.recv())
    pipe.send('hello, too')

# Build a pipe
pipe = mul.Pipe()
if __name__ == '__main__':
    # Pass an end of the pipe to process 1
    p1 = mul.Process(target=proc1, args=(pipe[0],))
    # Pass the other end of the pipe to process 2
    p2 = mul.Process(target=proc2, args=(pipe[1],))
    p1.start()
    p2.start()
    p1.join()
    p2.join()

# 结果打印
proc2 rec: hello
proc1 rec: hello, too
```

## Queue 队列

**Queue**与Pipe相类似，都是**先进先出的结构**。但**Queue允许多个进程放入，多个进程从队列取出对象**。Queue使用mutiprocessing.Queue(maxsize)创建，maxsize表示队列中可以存放对象的最大数量。

```
import os
import multiprocessing
import time
#==================
# input worker
def inputQ(queue):
    info = str(os.getpid()) + '(put):' + str(time.time())
    queue.put(info)

# output worker
def outputQ(queue,lock):
    info = queue.get()
    lock.acquire()
    print (str(os.getpid()) + ' get: ' + info)
    lock.release()
#===================
# Main
record1 = []   # store input processes
record2 = []   # store output processes
lock  = multiprocessing.Lock()    # To prevent messy print
queue = multiprocessing.Queue(3)

if __name__ == '__main__':
    # input processes
    for i in range(10):
        process = multiprocessing.Process(target=inputQ,args=(queue,))
        process.start()
        record1.append(process)
    
    # output processes
    for i in range(10):
        process = multiprocessing.Process(target=outputQ,args=(queue,lock))
        process.start()
        record2.append(process)
    
    for p in record1:
        p.join()
    
    queue.close()  # No more object will come, close the queue
    
    for p in record2:
        p.join()

# 打印结果
521278 get: 521262(put):1676452350.1724684
521275 get: 521266(put):1676452350.1747067
521279 get: 521267(put):1676452350.1759229
521276 get: 521268(put):1676452350.1771102
521283 get: 521263(put):1676452350.179253
521277 get: 521265(put):1676452350.180262
521281 get: 521272(put):1676452350.1832955
521282 get: 521270(put):1676452350.1812913
521280 get: 521274(put):1676452350.1853118
521284 get: 521273(put):1676452350.1842883

根据时间戳判断，符合先进先出原则
```

## 进程池 pool
**进程池 (Process Pool)可以创建多个进程**。这些**进程就像是随时待命的士兵，准备执行任务(程序)**。一个进程池中可以容纳多个待命的进程。

```
import multiprocessing as mul

def f(x):
    return x ** 2

if __name__ == '__main__':
    pool = mul.Pool(5)
    rel = pool.map(f, [1, 2, 3, 4, 5, 6, 7, 8, 9, 10])
    print(rel)

# 打印结果
[1, 4, 9, 16, 25, 36, 49, 64, 81, 100]
```

我们创建了一个容许5个进程的进程池 (Process Pool) 。Pool运行的每个进程都执行f()函数。我们利用map()方法，将f()函数作用到表的每个元素上。这与built-in的map()函数类似，只是这里用5个进程并行处理。如果进程运行结束后，还有需要处理的元素，那么的进程会被用于重新运行f()函数。

**除了map()方法外，Pool还有下面的常用方法**：
- apply_async(func,args) 从进程池中取出一个进程执行func，args为func的参数。它将返回一个AsyncResult的对象，你可以对该对象调用get()方法以获得结果。
- close() 进程池不再创建新的进程
- join() wait进程池中的全部进程。必须对Pool先调用close()方法才能join。

## 共享内存
### Value、Array
```
import multiprocessing

# Value/Array
def func1(a, arr):
    a.value = 3.14
    for i in range(len(arr)):
        arr[i] = 0
    a.value = 0

if __name__ == '__main__':
    num = multiprocessing.Value('d', 1.0)  # num=0
    arr = multiprocessing.Array('i', range(10))  # arr=range(10)
    p = multiprocessing.Process(target=func1, args=(num, arr))
    p.start()
    p.join()
    print (num.value)
    print (arr[:])

# 打印结果
0.0
[0, 0, 0, 0, 0, 0, 0, 0, 0, 0]
```

这里我们实际上只有主进程和Process对象代表的进程。我们在主进程的内存空间中创建共享的内存，也就是Value和Array两个对象。对象Value被设置成为双精度数(d), 并初始化为1.0。而Array则类似于C中的数组，有固定的类型(i, 也就是整数)。在Process进程中，我们修改了Value和Array对象。回到主程序，打印出结果，主程序也看到了两个对象的改变，说明资源确实在两个进程之间共享。

### manager
Manager是通过共享进程的方式共享数据。
Manager管理的共享数据类型有：Value、Array、dict、list、Lock、Semaphore等等，同时Manager还可以共享类的实例对象。

实例代码:
```
from multiprocessing import Process,Manager
def func1(shareList,shareValue,shareDict,lock):
    with lock:
        shareValue.value+=1
        shareDict[1]='1'
        shareDict[2]='2'
        for i in xrange(len(shareList)):
            shareList[i]+=1

if __name__ == '__main__':
    manager=Manager()
    list1=manager.list([1,2,3,4,5])
    dict1=manager.dict()
    array1=manager.Array('i',range(10))
    value1=manager.Value('i',1)
    lock=manager.Lock()
    proc=[Process(target=func1,args=(list1,value1,dict1,lock)) for i in xrange(20)]
    for p in proc:
        p.start()
    for p in proc:
        p.join()
    print list1
    print dict1
    print array1
    print value1

# 打印结果
```

# 多线程模块 threading
Python3 线程中常用的两个模块为：
- _thread
- threading(推荐使用)
thread 模块已被废弃。用户可以使用 threading 模块代替。所以，在 Python3 中不能再使用"thread" 模块。为了兼容性，Python3 将 thread 重命名为 "_thread"。

## threading模块
- threading 模块除了包含 _thread 模块中的所有方法外，还提供的其他方法：
- threading.currentThread(): 返回当前的线程变量。
- threading.enumerate(): 返回一个包含正在运行的线程的list。正在运行指线程启动后、结束前，不包括启动前和终止后的线程。
- threading.activeCount(): 返回正在运行的线程数量，与len(threading.enumerate())有相同的结果。

### Thread 类
除了使用方法外，线程模块同样提供了Thread类来处理线程，Thread类提供了以下方法:
- run(): 用以表示线程活动的方法。
- start():启动线程活动。
- join([time]): 等待至线程中止。这阻塞调用线程直至线程的join() 方法被调用中止-正常退出或者抛出未处理的异常-或者是可选的超时发生。
- isAlive(): 返回线程是否活动的。
- getName(): 返回线程名。
- setName(): 设置线程名。

## 示例
我们可以通过直接从 threading.Thread 继承创建一个新的子类，并实例化后调用 start() 方法启动新线程，即它调用了线程的 run() 方法：
```
import threading
import time

exitFlag = 0

class myThread (threading.Thread):
    def __init__(self, threadID, name, delay):
        threading.Thread.__init__(self)
        self.threadID = threadID
        self.name = name
        self.delay = delay
    def run(self):
        print ("开始线程：" + self.name)
        print_time(self.name, self.delay, 5)
        print ("退出线程：" + self.name)

def print_time(threadName, delay, counter):
    while counter:
        if exitFlag:
            threadName.exit()
        time.sleep(delay)
        print ("%s: %s" % (threadName, time.ctime(time.time())))
        counter -= 1

# 创建新线程
thread1 = myThread(1, "Thread-1", 1)
thread2 = myThread(2, "Thread-2", 2)

# 开启新线程
thread1.start()
thread2.start()
thread1.join()
thread2.join()
print ("退出主线程")

# 结果打印
开始线程：Thread-1
开始线程：Thread-2
Thread-1: Wed Feb 15 15:55:12 2023
Thread-2: Wed Feb 15 15:55:13 2023
Thread-1: Wed Feb 15 15:55:13 2023
Thread-1: Wed Feb 15 15:55:14 2023
Thread-2: Wed Feb 15 15:55:15 2023
Thread-1: Wed Feb 15 15:55:15 2023
Thread-1: Wed Feb 15 15:55:16 2023
退出线程：Thread-1
Thread-2: Wed Feb 15 15:55:17 2023
Thread-2: Wed Feb 15 15:55:19 2023
Thread-2: Wed Feb 15 15:55:21 2023
退出线程：Thread-2
退出主线程
```

## 线程同步（线程锁）
如果多个线程共同对某个数据修改，则可能出现不可预料的结果，为了保证数据的正确性，需要对多个线程进行同步。
使用 Thread 对象的 Lock 和 Rlock 可以实现简单的线程同步，这两个对象都有 acquire 方法和 release 方法，对于那些需要每次只允许一个线程操作的数据，可以将其操作放到 acquire 和 release 方法之间。

```
import threading
import time

class myThread (threading.Thread):
    def __init__(self, threadID, name, delay):
        threading.Thread.__init__(self)
        self.threadID = threadID
        self.name = name
        self.delay = delay
    def run(self):
        print ("开启线程： " + self.name)
        # 获取锁，用于线程同步
        threadLock.acquire()
        print_time(self.name, self.delay, 3)
        # 释放锁，开启下一个线程
        threadLock.release()

def print_time(threadName, delay, counter):
    while counter:
        time.sleep(delay)
        print ("%s: %s" % (threadName, time.ctime(time.time())))
        counter -= 1

threadLock = threading.Lock()
threads = []

# 创建新线程
thread1 = myThread(1, "Thread-1", 1)
thread2 = myThread(2, "Thread-2", 2)

# 开启新线程
thread1.start()
thread2.start()

# 添加线程到线程列表
threads.append(thread1)
threads.append(thread2)

# 等待所有线程完成
for t in threads:
    t.join()
print ("退出主线程")

# 结果打印
开启线程： Thread-1
开启线程： Thread-2
Thread-1: Wed Feb 15 16:02:00 2023
Thread-1: Wed Feb 15 16:02:01 2023
Thread-1: Wed Feb 15 16:02:02 2023
Thread-2: Wed Feb 15 16:02:04 2023
Thread-2: Wed Feb 15 16:02:06 2023
Thread-2: Wed Feb 15 16:02:08 2023
退出主线程
```

## 优先级队列（ Queue）
Python 的 Queue 模块中提供了同步的、线程安全的队列类，包括FIFO（先入先出)队列Queue，LIFO（后入先出）队列LifoQueue，和优先级队列 PriorityQueue。
这些队列都实现了锁原语，能够在多线程中直接使用，可以使用队列来实现线程间的同步。
- Queue 模块中的常用方法:
- Queue.qsize() 返回队列的大小
- Queue.empty() 如果队列为空，返回True,反之False
- Queue.full() 如果队列满了，返回True,反之False
- Queue.full 与 maxsize 大小对应
- Queue.get([block[, timeout]])获取队列，timeout等待时间
- Queue.get_nowait() 相当Queue.get(False)
- Queue.put(item) 写入队列，timeout等待时间
- Queue.put_nowait(item) 相当Queue.put(item, False)
- Queue.task_done() 在完成一项工作之后，Queue.task_done()函数向任务已经完成的队列发送一个信号
- Queue.join() 实际上意味着等到队列为空，再执行别的操作

```
import queue
import threading
import time

exitFlag = 0

class myThread (threading.Thread):
    def __init__(self, threadID, name, q):
        threading.Thread.__init__(self)
        self.threadID = threadID
        self.name = name
        self.q = q

def run(self):
        print ("开启线程：" + self.name)
        process_data(self.name, self.q)
        print ("退出线程：" + self.name)

def process_data(threadName, q):
    while not exitFlag:
        queueLock.acquire()
        if not workQueue.empty():
            data = q.get()
            queueLock.release()
            print ("%s processing %s" % (threadName, data))
        else:
            queueLock.release()
        time.sleep(1)

threadList = ["Thread-1", "Thread-2", "Thread-3"]
nameList = ["One", "Two", "Three", "Four", "Five"]
queueLock = threading.Lock()
workQueue = queue.Queue(10)
threads = []
threadID = 1

# 创建新线程
for tName in threadList:
    thread = myThread(threadID, tName, workQueue)
    thread.start()
    threads.append(thread)
    threadID += 1

# 填充队列
queueLock.acquire()
for word in nameList:
    workQueue.put(word)
queueLock.release()

# 等待队列清空
while not workQueue.empty():
    pass

# 通知线程是时候退出
exitFlag = 1

# 等待所有线程完成
for t in threads:
    t.join()
print ("退出主线程")

# 结果打印
开启线程：Thread-1
开启线程：Thread-2
开启线程：Thread-3
Thread-2 processing One
Thread-3 processing Two
Thread-1 processing Three
Thread-1 processing Four
Thread-3 processing Five
退出线程：Thread-3
退出线程：Thread-2
退出线程：Thread-1
退出主线程
```

上一篇：Python 匿名函数下一篇：Python 日志模块 logging

你觉得文章怎么样