python 复习—并发编程实战——多线程和多进程的生产者消费者模型线程进程再总结

Posted 2021-09-26 胖虎是只mao

任务

生产者负责生产数据（随机生成一个数字），消费者负责处理数据（判断生产出来的数据是不是质数）。

现在假设有一个生产者和多个消费者。

1. 多线程（Thread）的生产者消费者模型

import random
import time
import math
import threading
from queue import Queue

class Producer(threading.Thread): # 生产者线程
    def __init__(self, data, num_sum):
        super(Producer, self).__init__()
        self.data = data
        self.num_sum = num_sum

    def run(self):
        for idx in range(self.num_sum):
            self.data.put(random.randint(0,999999999)) # 队列进


class Consumer(threading.Thread): # 消费者线程
    def __init__(self, data, result, num_sum):
        super(Consumer, self).__init__()
        self.data = data
        self.result = result
        self.num_sum = num_sum
        self.daemon = True # 当主进程结束的时候，子进程也结束

    def judger(self, n):
        if n < 2: return False
        for i in range(2,int(math.sqrt(n))+1):
            if n % i == 0:
                return False
        return True

    def run(self):   
        while(len(self.result)<self.num_sum): # 线程里面尽量不要用 while True
            try:
                number = self.data.get(False) #队列出（不等待）
                self.judger(number)
                self.result.append(number)
            except:
                if len(self.result)==self.num_sum:
                    print(len(self.result))
        print('Thread end')
        return

if __name__ == "__main__":
    data = Queue() # 队列
    threads = [] # 进程组
    result = [] # 结果存储
    num_sum = 1000000 # 处理任务的数量
    threads_limit = 4 # 开设的消费者线程数
    start = time.clock()
    producer = Producer(data, num_sum)
    producer.start()
    threads.append(producer)
    for i in range(threads_limit):
        consumer = Consumer(data, result, num_sum)
        consumer.start()
        threads.append(consumer)

    for t in threads:
        t.join()
    print(len(result))
    print('using {} threads cost time {}s'.format(threads_limit,time.clock()-start))

运行结果 ：

1个消费者线程：

1000000
using 1 threads cost time 117.2472726s

2个消费者线程：

1000000
using 2 threads cost time 110.3796141s

4个消费者线程：

1000000
using 4 threads cost time 170.1167697s

2. 多进程（Process）的生产者消费者模型

import random
import time
import math
from multiprocessing import Queue, JoinableQueue, Process
import multiprocessing
import os


class Producer(Process): # 生产者进程
    def __init__(self, data, num_sum):
        super(Producer, self).__init__()
        self.data = data
        self.num_sum = num_sum

    def run(self):
        for idx in range(self.num_sum):
                self.data.put(random.randint(0,999999999)) # 队列进


class Consumer(Process): # 消费者进程
    def __init__(self, data, result, num_sum):
        super(Consumer, self).__init__()
        self.data = data
        self.result = result
        self.num_sum = num_sum

    def judger(self, n):
        if n < 2: return False
        for i in range(2,int(math.sqrt(n))+1):
            if n % i == 0:
                return False
        return True

    def run(self):   
        while(self.result.qsize()<self.num_sum):
            try:
                number = self.data.get(False) #队列出（不等待）
                self.judger(number)
                self.result.put(number)
            except:
                if self.result.qsize()==self.num_sum:
                    print(self.result.qsize())
                    
        print('Process end')
        os._exit(0) # You must need this
        return

if __name__ == "__main__":
    data = Queue()
    processes = [] # 进程组
    result = Queue()
    num_sum = 1000000 # 处理任务的数量
    processes_limit = 4 # 开设的消费者进程数
    start = time.clock()
    producer = Producer(data, num_sum)
    producer.start()
    processes.append(producer)

    for i in range(processes_limit):
        consumer = Consumer(data,result,num_sum)
        consumer.start()
        processes.append(consumer)

    for p in processes:
        p.join()

    print(result.qsize())
    print('using {} processes cost time {}s'.format(processes_limit,time.clock()-start))

1个消费者进程：

1000000
using 1 processes cost time 120.4088335s

2个消费者进程：

1000000
using 2 processes cost time 77.3745486s

4个消费者进程：

1000000
using 4 processes cost time 69.52899710000001s

3. 实验结论：

明显可见：上面的多进程比单进程明显更快，多线程和单线程的处理速度类似。

3.1 多线程和多进程

多线程：在同一个进程，CPU使用同一片的内存，共享同一份的全局变量，共享内存资源，来执行不同的任务，这叫并发。
多进程：不同的进程在不同的内存区域，不共享全局变量，不共享内存资源，但共享CPU的运算资源（时间片），这叫并行。

3.2 Python多线程多进程要点总结

在多线程中，不同线程对于主线程的全局资源的访问是自由的，我们可以使用队列类 Queue 来实现生产者消费者共享资源的缓冲区
在多进程中，不同的进程不能自由访问主线程的全局资源，但是我们可以使用multiprocessing的 Queue 来实现生产者消费者共享资源的缓冲区

多进程完成不一定会自动结束，因此在某些情况下我们需要手动结束，使用os._exit(0)。需要注意的是，有时候主进程结束了，子进程不一定结束的，这个情况很怪异，我们想如果正常的是主进程在结束的时候会向子进程发送信号告诉他结束，但是我们却不能百分百保证到的是主进程在发送子进程结束信号之前就不挂掉，万一出现这样的情况，我们把这种子进程叫做“孤儿进程”，因此，我们的子进程需要有自我结束的处理块。
如果我们需要实现线程/进程同步的话，我们需要同步线程/进程，在需要同步的代码之前执行 join()，前提是我们的这些进程/线程都start()起来了。

subprocess 和 multiprocessing 的区别在于：sub主要是面向是用在调用外部进程，multi主要是面向用在调用自身代码上的进程

以上提到的两种 Queue 是不一样的，一个是用来线程通信的，一个用来进程通信的

3.4 讨论：什么时候用多线程，什么时候用多进程？

ps： 可靠性应该是一个进程挂掉导致整个线程挂掉

1）需要频繁创建销毁的优先用线程

原因请看上面的对比。

这种原则最常见的应用就是Web服务器了，来一个连接建立一个线程，断了就销毁线程，要是用进程，创建和销毁的代价是很难承受的

3）强相关的处理用线程，弱相关的处理用进程

什么叫强相关、弱相关？理论上很难定义，给个简单的例子就明白了。

一般的Server需要完成如下任务：消息收发、消息处理。“消息收发”和“消息处理”就是弱相关的任务，而“消息处理”里面可能又分为“消息解码”、“业务处理”，这两个任务相对来说相关性就要强多了。因此“消息收发”和“消息处理”可以分进程设计，“消息解码”、“业务处理”可以分线程设计。

当然这种划分方式不是一成不变的，也可以根据实际情况进行调整。

4）可能要扩展到多机分布的用进程，多核分布的用线程

原因请看上面对比。

5）都满足需求的情况下，用你最熟悉、最拿手的方式

至于“数据共享、同步”、“编程、调试”、“可靠性”这几个维度的所谓的“复杂、简单”应该怎么取舍，我只能说：没有明确的选择方法。但我可以告诉你一个选择原则：如果多进程和多线程都能够满足要求，那么选择你最熟悉、最拿手的那个。

需要提醒的是：虽然我给了这么多的选择原则，但实际应用中基本上都是“进程+线程”的结合方式，千万不要真的陷入一种非此即彼的误区。

消耗资源：

从内核的观点看，进程的目的就是担当分配系统资源（CPU时间、内存等）的基本单位。线程是进程的一个执行流，是CPU调度和分派的基本单位，它是比进程更小的能独立运行的基本单位。

线程，它们彼此之间使用相同的地址空间，共享大部分数据，启动一个线程所花费的空间远远小于启动一个进程所花费的空间，而且，线程间彼此切换所需的时间也远远小于进程间切换所需要的时间。据统计，总的说来，一个进程的开销大约是一个线程开销的30倍左右，当然，在具体的系统上，这个数据可能会有较大的区别。

通讯方式：

进程之间传递数据只能是通过通讯的方式，即费时又不方便。线程时间数据大部分共享（线程函数内部不共享），快捷方便。但是数据同步需要锁对于static变量尤其注意

线程自身优势：

提高应用程序响应；使多CPU系统更加有效。操作系统会保证当线程数不大于CPU数目时，不同的线程运行于不同的CPU上；

改善程序结构。一个既长又复杂的进程可以考虑分为多个线程，成为几个独立或半独立的运行部分，这样的程序会利于理解和修改