并发编程

概念
    # 同步异步阻塞和非阻塞
        # 同步阻塞 : 调用一个函数需要等待这个函数的执行结果,并且在执行这个函数的过程中CPU不工作
            # inp = input('>>>')
        # 同步非阻塞 :调用一个函数需要等待这个函数的执行结果,在执行这个函数的过程中CPU工作
            # ret = eval('1+2+3-4')
        # 异步非阻塞 :调用一个函数不需要等待这个函数的执行结果,并且在执行这个函数的过程中CPU工作
            # start()
        # 异步阻塞 : 调用一个函数不需要等待这个函数的执行结果,并且在执行这个函数的过程中CPU不工作
            # 开启10个进程 异步的
            # 获取这个进程的返回值,并且能做到哪一个进程先结束,就先获取谁的返回值
    # 进程的三状态图
        # 就绪 -操作系统调度->运行 -遇到io操作-> 阻塞 -阻塞状态结束-> 就绪
        #                         -时间片到了-> 就绪
    # 进程的调度算法 : 短作业和长作业是有区别的,越长的作业被调度的没有短作业调度的积极
                     # 每一个io操作都会让你辛苦排来的队执行的CPU机会让给其他程序
        # 先来先服务
        # 短作业优先
        # 分时的概念
        # 多级反馈算法
    # 进程开启和关闭
        # 父进程 开启了 子进程
        # 父进程 要负责给 子进程 回收子进程结束之后的资源

1.multiprocessing 多元的处理进程的模块

import os
from multiprocessing import Process

print(os.getpid(), os.getppid())  #进程ID1828    父进程id 3852

def func():
    print(os.getpid(),os.getppid())
    # pid process id           进程id
    # ppid parent process id   父进程id



if __name__ == '__main__':
    # 只会在主进程中执行的所有的代码你写在name = main下
    print('main :',os.getpid(),os.getppid())
    p = Process(target=func)
    p.start()         #开启一个子进程

能不能给子进程传递参数   
#能  p = Process(target=func,args=('alex',84))

# import os
# from multiprocessing import Process
#
# def func(name,age):
#     print(os.getpid(),os.getppid(),name,age)
#
# if __name__ == '__main__':
#     # 只会在主进程中执行的所有的代码你写在name = main下
#     print('main :',os.getpid(),os.getppid())
#     p = Process(target=func,args=('alex',84))
#     p.start()

能不能获取子进程的返回值   # 不能

# 能不能同时开启多个子进程  可以

import os
import time
from multiprocessing import Process

def func(name,age):
    print('%s start'%name)
    time.sleep(1)        #异步并发
    print(os.getpid(),os.getppid(),name,age)

if __name__ == '__main__':
    # 只会在主进程中执行的所有的代码你写在name = main下
    print('main :',os.getpid(),os.getppid())
    arg_lst = [('alex',84),('太白', 40),('wusir', 48)]
    for arg in arg_lst:
        p = Process(target=func,args=arg)  #args接收的参数必须是元组或列表
        p.start()  # 异步非阻塞

2.join的用法

import os
import time
import random
import  json

from multiprocessing import Process



def func(name,age):
    print('发送一封邮件给%s岁的%s'%(age,name))
    # time.sleep(1)
    time.sleep(random.random())  #随机休眠
    print('发送完毕')

if __name__ =='__main__':
    arg_lst = [('大壮',40),('alex', 84), ('太白', 40), ('wusir', 48)]
    p_list = []

    for i in arg_lst:
        p = Process(target=func, args=i)
        p.start()
        p_list.append(i)
    for p in p_list:
        p.join()
        # p.join()# 阻塞：直到p这个子进程执行完毕才继续执行    这是同步阻塞
    print('打印完成')

# 同步阻塞 异步非阻塞
    # 同步阻塞 join
    # 异步非阻塞 start

多进程之间的数据是否隔离   隔离的

from multiprocessing import Process
n = 0
def func():
    global n
    n += 1

if __name__ == '__main__':
    p_l = []
    for i in range(100):
        p = Process(target=func)
        p.start()
        p_l.append(p)
    for p in p_l:p.join()
    print(n)

同步异步阻塞非阻塞

　在了解其他概念之前，我们首先要了解进程的几个状态。在程序运行的过程中，由于被操作系统的调度算法控制，程序会进入几个状态：就绪，运行和阻塞。

　　（1）就绪(Ready)状态

　　当进程已分配到除CPU以外的所有必要的资源，只要获得处理机便可立即执行，这时的进程状态称为就绪状态。

　　（2）执行/运行（Running）状态当进程已获得处理机，其程序正在处理机上执行，此时的进程状态称为执行状态。

　　（3）阻塞(Blocked)状态正在执行的进程，由于等待某个事件发生而无法执行时，便放弃处理机而处于阻塞状态。引起进程阻塞的事件可有多种，例如，等待I/O完成、申请缓冲区不能满足、等待信件(信号)等。

      

process模块介绍

process模块是一个创建进程的模块，借助这个模块，就可以完成进程的创建。

Process([group [, target [, name [, args [, kwargs]]]]])，由该类实例化得到的对象，表示一个子进程中的任务（尚未启动）

强调：
1. 需要使用关键字的方式来指定参数
2. args指定的为传给target函数的位置参数，是一个元组形式，必须有逗号

参数介绍：
1 group参数未使用，值始终为None
2 target表示调用对象，即子进程要执行的任务
3 args表示调用对象的位置参数元组，args=(1,2,'egon',)
4 kwargs表示调用对象的字典,kwargs={'name':'egon','age':18}
5 name为子进程的名称

1 p.start()：启动进程，并调用该子进程中的p.run() 
2 p.run():进程启动时运行的方法，正是它去调用target指定的函数，我们自定义类的类中一定要实现该方法  
3 p.terminate():强制终止进程p，不会进行任何清理操作，如果p创建了子进程，该子进程就成了僵尸进程，使用该方法需要特别小心这种情况。如果p还保存了一个锁那么也将不会被释放，进而导致死锁
4 p.is_alive():如果p仍然运行，返回True
5 p.join([timeout]):主线程等待p终止（强调：是主线程处于等的状态，而p是处于运行的状态）。timeout是可选的超时时间，需要强调的是，p.join只能join住start开启的进程，而不能join住run开启的进程  

1 p.daemon：默认值为False，如果设为True，代表p为后台运行的守护进程，当p的父进程终止时，p也随之终止，并且设定为True后，p不能创建自己的新进程，必须在p.start()之前设置
2 p.name:进程的名称
3 p.pid：进程的pid
4 p.exitcode:进程在运行时为None、如果为–N，表示被信号N结束(了解即可)
5 p.authkey:进程的身份验证键,默认是由os.urandom()随机生成的32字符的字符串。这个键的用途是为涉及网络连接的底层进程间通信提供安全性，这类连接只有在具有相同的身份验证键时才能成功（了解即可）

在windoes使用时要注意的事项

在Windows操作系统中由于没有fork(linux操作系统中创建进程的机制)，在创建子进程的时候会自动 import 启动它的这个文件，而在 import 的时候又执行了整个文件。因此如果将process()直接写在文件中就会无限递归创建子进程报错。所以必须把创建子进程的部分使用if __name__ ==‘__main__’ 判断保护起来，import 的时候  ，就不会递归运行了。

守护进程

会随着主进程的结束而结束。

主进程创建守护进程

　　其一：守护进程会在主进程代码执行结束后就终止

　　其二：守护进程内无法再开启子进程,否则抛出异常：AssertionError: daemonic processes are not allowed to have children

注意：进程之间是互相独立的，主进程代码运行结束，守护进程随即终止

1.守护进程的启动

import os
import time
from multiprocessing import Process

class Myprocess(Process):
    def __init__(self,person):
        super().__init__()
        self.person = person
    def run(self):
        print(os.getpid(),self.name)
        print('%s正在和女主播聊天' %self.person)


p=Myprocess('哪吒')
p.daemon=True #一定要在p.start()前设置,设置p为守护进程,禁止p创建子进程,并且父进程代码执行结束,p即终止运行
p.start()
time.sleep(10) # 在sleep时查看进程id对应的进程ps -ef|grep id
print('主')

2.主进程代码执行结束,守护进程立即结束

from multiprocessing import Process

def foo():
    print(123)
    time.sleep(1)
    print("end123")

def bar():
    print(456)
    time.sleep(3)
    print("end456")


p1=Process(target=foo)
p2=Process(target=bar)

p1.daemon=True
p1.start()
p2.start()
time.sleep(0.1)
print("main-------")#打印该行则主进程代码结束,则守护进程p1应该被终止.#可能会有p1任务执行的打印信息123,因为主进程打印main----时,p1也执行了,但是随即被终止.

3.Process类的其它方法

from multiprocessing import Process
import os
import time

class Myprocess(Process):
    def __init__(self,a,b,c):
        self.a = a
        self.b = b
        self.c = c
        super().__init__()


    def run(self):
        time.sleep(1)
        print(os.getpid(),os.getppid(),self.a,self.b,self.c)

if __name__ == '__main__':
    p = Myprocess(1,2,3)
    p.start()
    print(p.pid,p.ident)  #查看子进程ID
    print(p.name) #查看进程名字
    print(p.is_alive()) #查看进程是否存在
    p.terminate() #强制杀死进程
    time.sleep(1)
    print(p.is_alive())

守护进程   要求守护进程p1必须在p2进程执行结束之后才结束

from multiprocessing import  Process

import time
import os

# 要求守护进程p1必须在p2进程执行结束之后才结束
def son1():
    while True:
        print('<<<in son1')
        time.sleep(1)

def son2():
    for i in range(10):
        print('in son2')
        time.sleep(1)

if __name__ == '__main__':
    p = Process(target=son1)
    p.daemon = True
    p.start()
    p2 = Process(target=son2)
    p2.start()
    time.sleep(3)
    print('in _main_')
    p2.join()

守护进程:锁

抢票的例子  一个人抢到了.另外的人就不能够抢了

import os
import json,time
from multiprocessing import Process,Lock


def search(i):
    with open('ticker',mode='r',encoding='utf-8') as f:
        ticker = json.load(f)
        print('%s:当前还剩于%s张票'%(i,ticker['count']))



def buy_ticker(i):
    with open('ticker',mode='r',encoding='utf-8') as f:
        ticker = json.load(f)
        if ticker['count'] > 0:
            ticker['count'] -= 1
            print('%s买到票了'%(i))
        time.sleep(0.1)
    with open('ticker',mode='w',encoding='utf-8') as f:
        json.dump(ticker,f)


def get_ticker(i,lock):
    search(i)

    # lock.cquire()  #拿钥匙
    # buy_ticker(i)
    # lock.release()   #还钥匙

    with lock:   #锁
        buy_ticker(i)



if __name__ == '__main__':
    lock = Lock()

    for i in range(10):
        p = Process(target=get_ticker,args=(i,lock))
        p.start()

队列------完善的生产消费者模型  *****

 # 把原本获取数据处理数据的完整过程进行了 解耦
    # 把生产数据和消费数据分开,根据生产和消费的效率不同,来规划生产者和消费者的个数,
       # 让程序的执行效率达到平衡
# 如果你写了一个程序所有的功能代码都放在一起,不分函数不分类也不分文件
# 紧耦合的程序

# 拆分的很清楚的程序
# 松耦合的程序

# 进程的生产者消费者模型(默写)
    # 多个进程访问网页
    # 一个进程负责把网页源码写到文件里
    # 消费者如何结束
    # 哪些部分是生产者消费者模型比较重要的结构

import time
import random
from multiprocessing import Queue,Process


def consumer(q,name): # 消费者:通常取到数据之后还要进行某些操作
    while True:
        food = q.get()
        if food:
            print('%s吃了%s'%(name,food))
        else:break

def producer(q,name,food): # 生产者:通常在放数据之前需要先通过某些代码来获取数据
    for i in range(10):
        foodi = '%s%s'%(food,i)
        print('%s生产了%s'%(name,foodi))
        time.sleep(random.random())
        q.put(foodi)

if __name__ == '__main__':
    q = Queue()
    c1 = Process(target=consumer,args=(q,'alex'))
    c2 = Process(target=consumer,args=(q,'alex'))
    p1 = Process(target=producer,args=(q,'大壮','泔水'))
    p2 = Process(target=producer,args=(q,'b哥','香蕉'))
    c1.start()
    c2.start()
    p1.start()
    p2.start()
    p1.join()
    p2.join()
    q.put(None)  #有多少个消费者.就需要有多少个put(none)
    q.put(None)

同步阻塞
    # 调用函数必须等待结果cpu没工作 input sleep recv accept connect get
# 同步非阻塞
    # 调用函数必须等待结果cpu工作 - 调用了一个高计算的函数 strip eval('1+2+3') sum max min sorted
# 异步阻塞
    # 调用函数不需要立即获取结果,而是继续做其他的事情,在获取结果的时候不知道先获取谁的,但是总之需要等(阻塞)
# 异步非阻塞
    #  调用函数不需要立即获取结果,也不需要等 start() terminate()

异步阻塞


import requests
from multiprocessing import Process,Queue
url_dic = {
    'cnblogs':'https://www.cnblogs.com/Eva-J/articles/8253549.html',
    'douban':'https://www.douban.com/doulist/1596699/',
    'baidu':'https://www.baidu.com',
    'gitee':'https://gitee.com/old_boy_python_stack__22/teaching_plan/issues/IXSRZ',

}

def producer(name,url,q):
    ret = requests.get(url)
    q.put((name,ret.text))

def consumer(q):
    while True:
        tup = q.get()
        if tup is None:break
        with open('%s.html'%tup[0],encoding='utf-8',mode='w') as f:
            f.write(tup[1])

if __name__ == '__main__':
    q = Queue()
    pl = []
    for key in url_dic:
        p = Process(target=producer,args=(key,url_dic[key],q))
        p.start()
        pl.append(p)
    Process(target=consumer,args=(q,)).start()
    for p in pl:p.join()
    q.put(None)

数据共享  Manager 

# _*_ coding : UTF-8 _*_
from multiprocessing import Process,Manager,Lock

def change_dic(dic,lock):
    with lock:   #����  �а�ȫ����
        dic['count'] -= 1

if __name__ == '__main__':
    # m = Manager()    #�������ݹ���
    with Manager() as m:
        lock = Lock()
        dic = m.dict({'count': 100})
        # dic = {'count': 100}
        p_l = []
        for i in  range(100):
            p = Process(target=change_dic,args=(dic,lock))
            p.start()
            p_l.append(p)
        for p in p_l : p.join()
        print(dic)

 线程    threading模块开启线程

# 线程
    # 概念
        # 数据共享,效率高开销小,可以被多个cpu调度(是CPU调度的最小单位),数据不安全,由操作系统负责调度
        # 在cpython解释器下 :GIL锁(全局解释器锁) 导致了同一个进程中的多个线程不能利用多核
    # 代码
        # threading
            # 对象 = 实例化的结果 : 指定target args
                # start
                # join
                # ident
            # current_thread  能够帮助你获取当前这句函数所在的线程的线程对象
            # enumerate 导入之后会和内置函数enumerate重名,需要做特殊的处理
                # from threading import enumerate as en
                # import threading
                # threading.enumerate()
            # active_count 查看存活的线程个数(包括主线程)

threading模块 

from threading import Thread,current_thread,enumerate,active_count #线程

current_thread() 获取当前所在的线程的对象
enumerate 列表 存储了所有活着的线程对象,包括主线程
active_count 数字 存储了所有活着的线程个数

import os
import time
from threading import Thread,current_thread,enumerate,active_count    #线程

def func(i):
    print('start%s'%i,current_thread().ident)    #拿到所有线程的id。current_thread().是所有线程。ident是ID
    time.sleep(1)
    print('end%s'%i)
if __name__ == '__main__':
    tl = []
    for i in range(10):
        t = Thread(target=func,args=(i,))
        t.start()
        print(t.ident,os.getpid())
        tl.append(t)
    print(enumerate(),active_count())   #共11个线程。10个加一个主线程   active_count 是记录所有活着的线程个数
    for t in tl:t.join()
    print('所有的线程都执行完了')



# current_thread() 获取当前所在的线程的对象 current_thread().ident通过ident可以获取线程id
# 线程是不能从外部   terminate   terminate是关闭
# 所有的子线程只能是自己执行完代码之后就关闭
# enumerate 列表 存储了所有活着的线程对象,包括主线程
# active_count 数字 存储了所有活着的线程个数

# 面向对象的方式起线程


from threading import Thread
class MyThread(Thread):
    def __init__(self,a,b):
        self.a = a
        self.b = b
        super().__init__()

    def run(self):
        print(self.ident)

t = MyThread(1,2)
t.start()  # 开启线程 才在线程中执行run方法
print(t.ident)

# 线程之间的数据的共享


from threading import Thread
n = 100

def func():
    global n
    n -= 1

t_l = []
for i in range(100):
    t = Thread(target=func)
    t.start()
    t_l.append(t)
for t in t_l:
    t.join()
print(n)

守护线程

# 主线程会等待子线程结束之后才结束
# 为什么? # 主线程结束进程就会结束
# 守护线程随着主线程的结束而结束
# 守护线程会在主线程的代码结束之后继续守护其他子线程么? 会的

# 守护进程 会随着主进程的代码结束而结束
    # 如果主进程代码结束之后还有其他子进程在运行,守护进程不守护
# 守护线程 随着主线程的结束而结束
    # 如果主线程代码结束之后还有其他子线程在运行,守护线程也守护

# 为什么?
    # 守护进程和守护线程的结束原理不同
    # 守护进程需要主进程来回收资源
    # 守护线程是随着进程的结束才结束的
        # 其他子线程-->主线程结束-->主进程结束-->整个进程中所有的资源都被回收-->守护线程也会被回收

# 进程是资源分配单位
# 子进程都需要它的父进程来回收资源
# 线程是进程中的资源
# 所有的线程都会随着进程的结束而被回收的