并发编程02

子进程回收的两种方式

• join让主进程等待子进程结束后，并回收进程资源，主进程在结束并回收资源

from multiprocessing import Process
import time


def task():
    print('start...')
    time.sleep(3)
    print('end...')


if __name__ == '__main__':
    p_obj = Process(target=task)
    p_obj.start()

    p_obj.join()
    print('主进程结束....')

• 主进程正常结束，子进程与主进程一并被回收资源

from multiprocessing import Process
import time


def task():
    print('start...')
    time.sleep(3)
    print('end...')


if __name__ == '__main__':
    p_obj = Process(target=task)
    p_obj.start()
    print('主进程结束....')

僵尸进程

在子进程结束后，主进程没有正常结束，子进程的PID不会被回收

缺点：

​ ①：操作系统中的PID号都是有限的，如有子进程PID号无法正常回收，则会占用PID号

​ ②：PID号骂好了，则无法创建新的进程

​ ③：浪费资源

孤儿进程（没有坏处）

在子进程没有结束时，主进程没有正常结束，子进程PID不会被回收

孤儿院：操作系统优化机制

​ 当主进程意外终止，操作系统会检测是否有正在运行的子进程，会将他们放入孤儿院中，让操作系统帮你自动回收

守护进程

当主进程结束时，子进程必须结束（陪葬机制）

from multiprocessing import Process

import time


def dome(name):
    print(f'start....{name}')
    time.sleep(3)
    print(f'end...{name}')
    print('子进程结束...')


if __name__ == '__main__':
    p = Process(target=dome, args=('太监一号', ))

    # 守护进程必须在p.start()调用前设置
    p.daemon = True

    p.start()

    time.sleep(1)
    print('皇帝驾崩啦啊~~~~~')

# 打印结果：
# start....太监一号
# 皇帝驾崩啦啊~~~~~

进程间数据是隔离的

from multiprocessing import Process
import time

number = 10


def func():
    global number
    number = 100


def func1(number):
    number += 10


if __name__ == '__main__':
    p_obj = Process(target=func)
    p_obj1 = Process(target=func1, args=(number,))

    p_obj.start()
    p_obj1.start()
    p_obj.join()
    p_obj1.join()
    time.sleep(1)
    print(number)   # 10 ---> 证明数据是隔离的

进程互斥锁

进程之间数据不共享，但是共享同一套文件系统，所以访问同一个文件或同一个打印终端是没有问题的，而共享带来的就是竞争，竞争带来的结果就是错乱，如何控制，就是加锁处理

# 抢票例题
from multiprocessing import Process
from multiprocessing import Lock    # ---> 进程互斥锁
import random
import json
import time


# 抢票例子

# 1、查看余票
def search(name):
    # 1.读取data.json文件中的数据
    with open('data.json', 'r', encoding='utf-8')as f:
        data_dic = json.load(f)
        print(f'用户[{name}]查看余票，余票还剩[{data_dic.get("number")}]!')


# 2、若有余票，购买成功，票数会减少
def buy(name):
    # 网络延时
    with open('data.json', 'r', encoding='utf-8')as f:
        data_dic = json.load(f)

    # 进入这一步 证明最先抢到票
    if data_dic.get('number') > 0:
        data_dic['number'] -= 1
        time.sleep(random.randint(1, 3))
        with open('data.json', 'w', encoding='utf-8')as f:
            json.dump(data_dic, f)
        print(f'用户[{name}]，抢票成功！')

    else:
        print(f'用户[{name}]，抢票失败！')


def run(name, lock):
    search(name)

    lock.acquire()      # 加锁

    buy(name)

    lock.release()      # 释放锁


if __name__ == '__main__':
    lock = Lock()
    # 开启多进程，实现并发
    for line in range(10):
        p_obj = Process(target=run, args=(f'张全蛋{line}', lock))
        p_obj.start()

总结：加锁可以保证多个进程修改同一块数据时，同一时间只能有一个任务可以进行修改，即串行的修改，速度是慢了，但牺牲了效率却保证了数据安全，这就是互斥锁的特点

队列（遵循先进先出）

队列相当于一个第三方的管道，可以存放数据，基于队列，可以实现进程间的互相通信

创建队列有三种方式

from multiprocessing import Queue
from multiprocessing import JoinableQueue
import queue
# 第一种
# Queue(5)指的是队列中只能存放5份数据
q_obj = Queue(5)       # q_obj队列的对象
# 添加数据队列中
q_obj.put('张三')     # put:添加数据到队列中，将队列加满会阻塞
print('添第加1个')
q_obj.put('李四')
print('添加第2个')
q_obj.put('王二麻子')
print('添加第3个')
q_obj.put('张全蛋')
print('添加第4个')
q_obj.put('李小花')
print('添加第5个')
# q_obj.put('王小明')    # 到这里程序就会进入阻塞态
# print('添加第6个')

print(q_obj.get())      # 取出队列中的数据，取完再取会阻塞
print(q_obj.get())
print(q_obj.get())
print(q_obj.get())
print(q_obj.get())
# print(q_obj.get())    # 到这里程序就会进入阻塞态



# # 第二种
# JoinableQueue(5)指的是队列中只能存放5份数据
q_obj = JoinableQueue(5)       # q_obj队列的对象
# 添加数据队列中
q_obj.put('张三')     # put:添加数据到队列中，将队列加满会阻塞
print('添第加1个')
q_obj.put('李四')
print('添加第2个')
q_obj.put('王二麻子')
print('添加第3个')
q_obj.put('张全蛋')
print('添加第4个')
q_obj.put('李小花')
print('添加第5个')
# q_obj.put('王小明')    # 到这里程序就会进入阻塞态
# print('添加第6个')

print(q_obj.get())      # 取出队列中的数据，取完再取会阻塞
print(q_obj.get())
print(q_obj.get())
print(q_obj.get())
print(q_obj.get())
# print(q_obj.get())    # 到这里程序就会进入阻塞态



# # 第三种
# queue.Queue(5)指的是队列中只能存放5份数据
q_obj = queue.Queue(5)       # q_obj队列的对象
# 添加数据队列中
q_obj.put('张三')     # put:添加数据到队列中，将队列加满会阻塞
print('添第加1个')
q_obj.put('李四')
print('添加第2个')
q_obj.put('王二麻子')
print('添加第3个')
q_obj.put('张全蛋')
print('添加第4个')
q_obj.put('李小花')
print('添加第5个')
# q_obj.put('王小明')    # 到这里程序就会进入阻塞态
# print('添加第6个')

print(q_obj.get())      # 取出队列中的数据，取完再取会阻塞
print(q_obj.get())
print(q_obj.get())
print(q_obj.get())
print(q_obj.get())
# print(q_obj.get())    # 到这里程序就会进入阻塞态



# 还有put和get的进阶用法
q_obj.put_nowait('张三')     # put_nowait:添加数据到队列中，将队列加满会报错
print('添加第1个')
q_obj.put_nowait('李四')
print('添加第2个')
q_obj.put_nowait('王二麻子')
print('添加第3个')
q_obj.put_nowait('张全蛋')
print('添加第4个')
q_obj.put_nowait('李小花')
print('添加第5个')
q_obj.put_nowait('王小明')    # 到这里程序就会报错
print('添加第6个')

print(q_obj.get_nowait())      # 取出队列中的数据，取完再取会报错
print(q_obj.get_nowait())
print(q_obj.get_nowait())
print(q_obj.get_nowait())
print(q_obj.get_nowait())
print(q_obj.get_nowait())       # # 到这里程序就会报错

IPC机制（实现进程间通信）

通过队列实现进程间的通信

from multiprocessing import Process
from multiprocessing import JoinableQueue
import time


def task1(q):
    x = 100
    q.put(x)
    print('添加数据')
    time.sleep(3)
    print(q.get())


def task2(q):
    # 想要在task2中获取task1中的数据x
    res = q.get()
    print(f'获取的数据是[{res}]')
    q.put(9527)


if __name__ == '__main__':
    # 产生队列
    q = JoinableQueue(10)

    # 产生两个不同的子进程
    p1 = Process(target=task1, args=(q, ))
    p2 = Process(target=task2, args=(q, ))

    p1.start()
    p2.start()
    
# 打印结果：
# 添加数据
# 获取的数据是[100]
# 9527

生产者与消费者

生产者：生产数据的

消费者：使用数据的

解决供需不平衡问题

from multiprocessing import JoinableQueue
from multiprocessing import Process
import time


# 生产者：生产数据 ---> 队列
def producer(name, food, q):
    msg = f'{name}生产了{food}食物'
    # 生产一个食物，添加到队列中
    q.put(food)
    print(msg)


# 消费者：使用数据  <--- 队列
def customer(name, q):
    while True:
        try:
            time.sleep(0.5)
            # 若报错则跳出循环
            food = q.get_nowait()
            msg = f'{name}吃了{food}食物！'
            print(msg)

        except Exception:
            break


if __name__ == '__main__':
    q = JoinableQueue()

    # 创建生产者生产10个食物
    for line in range(10):
        p1 = Process(target=producer, args=('tank', f'Pig饲料{line}', q))
        p1.start()

    # 创建两个消费者
    c1 = Process(target=customer, args=('Pig1', q))
    c2 = Process(target=customer, args=('Pig2', q))

    c1.start()
    c2.start()

线程

什么是线程？

进程：资源单位

线程：执行单位

线程与进程都是虚拟的概念，为了更好表达某种事物

PS：开启一个进程一定会自带一个线程，线程才是真正的执行者

为什么要使用线程？

节省资源的占用

开启进程和开启线程的区别

​ 开启进程

​ ①：开辟一个名称空间,每开启一个进程都会占用一份内存资源

​ ②：会自带一个主线程

​ 开启线程

​ ①：一个进程可以开启多个线程，从进程的内存空间中申请执行单位

​ ②：节省内存资源

打个比方：

开启三个进程：占用三份内存空间

开启三个线程，从一个内存资源中，申请三个小的执行单位

PS：进程与进程之间数据都是隔离的，线程与线程之间的数据都是共享的

创建线程的两种方式

from multiprocessing import Process
from threading import Thread
import time
# 方式一
def task():
    print("线程start...")
    time.sleep(3)
    print("线程end...")


if __name__ == '__main__':
    p_obj = Process(target=task)
    p_obj.start()
    p_obj.join()
    print("进程...")


# # 方式二
class MyThread(Thread):
    def __init__(self, name):
        super().__init__()
        self.name = name

    def run(self) -> None:
        print("线程start...")
        time.sleep(3)
        print("线程end...")


if __name__ == '__main__':
    p = MyThread("线程")
    p.start()
    p.join()
    print("主线程")

线程互斥锁（和进程互斥锁一样使用）

from threading import Lock
from threading import Thread
import time
lock = Lock()

# 开启10个线程，对一个数据进行修改
number = 100


def task():
    global number

    lock.acquire()
    number2 = number
    # time.sleep(1)
    number = number2 - 1
    lock.release()


if __name__ == '__main__':

    list1 = []
    for line in range(10):
        t = Thread(target=task)
        t.start()
        list1.append(t)

    for t in list1:
        t.join()

    print(number)  # 90

线程池

他是基于concurrent.futures模块，用来限制创建的线程数量的

# 线程池限制线程数量
from concurrent.futures import ThreadPoolExecutor
pool = ThreadPoolExecutor(50)  # 限制线程数量为50


def task():
    import time
    time.sleep(1)


while True:
    pool.submit(task)  # 相当于Thread().start()

GIL全局解释器锁（纯理论）

GIL全局解释器锁，本质上就是一把互斥锁，保证数据安全

在Cpython解释器中，同一个进程下开启的多线程，同一时刻只能有一个线程执行，无法利用多核优势

GIL全局解释器锁的优缺点

优点：

​ 保证数据的安全

缺点：

​ 单个进程下，开启多个线程，牺牲执行效率，无法实现并行

注意：

IO密集型下使用多线程

计算密集型下使用多进程

协程

进程：资源单位

线程：执行单位

协程：单线程下实现并发

​ 在IO密集型情况下，使用协程能将效率提到最高

PS：协程不是任何单位，只是程序员YY出来的东西

PS：将效率提到极致的方法，多进程--->多线程--->让每个线程都实现协程

协程的目的：

手动实现“遇到IO切换 + 保存状态”去欺骗操作系统，让操作系统误以为没有IO操作，将CPU的执行权限给你

from gevent import monkey  # 猴子补丁
monkey.patch_all()  # 监听所有的任务是否有IO操作
from gevent import spawn  # spawn(任务)
from gevent import joinall
import time


def task1():
    print('start from task1...')
    time.sleep(1)
    print('end from task1...')


def task2():
    print('start from task2...')
    time.sleep(3)
    print('end from task2...')


def task3():
    print('start from task3...')
    time.sleep(5)
    print('end from task3...')


if __name__ == '__main__':

    start_time = time.time()
    sp1 = spawn(task1)
    sp2 = spawn(task2)
    sp3 = spawn(task3)

    joinall([sp1, sp2, sp3])

    end_time = time.time()

    print(f'消耗时间: {end_time - start_time}')

# 打印结果：
# start from task1...
# start from task2...
# start from task3...
# end from task1...
# end from task2...
# end from task3...
# 消耗时间: 5.02597451210022