Python基础之（并发编程）

一、进程

1.1、什么是进程

进程（Process）是计算机中的程序关于某数据集合上的一次运行活动，是系统进行资源分配和调度的基本单位，是操作系统结构的基础。在早期面向进程设计的计算机结构中，进程是程序的基本执行实体；在当代面向线程设计的计算机结构中，进程是线程的容器。程序是指令、数据及其组织形式的描述，而进程是程序的实体。

狭义定义：进程是正在运行的程序的实例

广义定义：进程是一个具有一定独立功能的程序关于某个数据集合的一次运行活动。它是操作系统动态执行的基本单元，在传统的操作系统中，进程既是基本的分配单元，也是基本的执行单元

1.2、并发与并行

 并行是指两者同时执行，比如赛跑，两个人都在不停的往前跑；（资源够用，比如三个线程，四核的CPU ）

 并发是指资源有限的情况下，两者交替轮流使用资源，比如一段路(单核CPU资源)同时只能过一个人，A走一段后，让给B，B用完继续给A ，交替使用，目的是提高效率（一种是伪并行，cpu多道技术IO切换；另一种就是并行，具备多个cpu）

1.3、同步、异步，阻塞、非阻塞

所谓同步就是一个任务的完成需要依赖另外一个任务时，只有等待被依赖的任务完成后，依赖的任务才能算完成，这是一种可靠的任务序列。要么成功都成功，失败都失败，两个任务的状态可以保持一致；异步是指不需要等待被依赖的任务完成，只是通知被依赖的任务要完成什么工作，依赖的任务也立即执行，只要自己完成了整个任务就算完成了。至于被依赖的任务最终是否真正完成，依赖它的任务无法确定，所以它是不可靠的任务序列。

阻塞是指调用结果返回之前，当前线程会被挂起（如遇到io操作）。函数只有在得到结果之后才会将阻塞的线程激活；非阻塞和阻塞的概念相对应，指在不能立刻得到结果之前也会立刻返回，同时该函数不会阻塞当前线程

# 同步与异步针对的是函数/任务的调用方式：同步就是当一个进程发起一个函数（任务）调用的时候，一直等到函数（任务）完成，而进程继续处于激活状态。而异步情况下是当一个进程发起一个函数（任务）调用的时候，不会等函数返回，而是继续往下执行当，函数返回的时候通过状态、通知、事件等方式通知进程任务完成。

# 阻塞与非阻塞针对的是进程或线程：阻塞是当请求不能满足的时候就将进程挂起，而非阻塞则不会阻塞当前进程

1.4、multiprocessing模块介绍

python中的多线程无法利用多核优势，如果想要充分地使用多核CPU的资源，在python中大部分情况需要使用多进程。Python提供了multiprocessing模块用于多进程。

multiprocessing模块用来开启子进程，并在子进程中执行我们定制的任务，该模块与多线程模块threading的编程接口类似。 multiprocessing模块的功能众多：支持子进程、通信和共享数据、执行不同形式的同步，提供了Process、Queue、Pipe、Lock等组件。

1.4.1、Process使用

from multiprocessing import Process
import time
def fun(name):
    print("hi",name)

if __name__ == "__main__":
    pro=Process(target=fun,args=("jump",))
    pro.start()
    time.sleep(2)
    print("hi,crazy")

参数介绍：

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字符的字符串。这个键的用途是为涉及网络连接的底层进程间通信提供安全性，这类连接只有在具有相同的身份验证键时才能成功（了解即可）

除了上面这些开启进程的方法，还有一种以继承Process类的形式开启进程的方式：

from multiprocessing import Process
class MyProcess(Process):
    def __init__(self,name):
        super().__init__()
        self.name=name
    def run(self):
        print('hi,%s' %self.name)

if __name__ == "__main__":
    p1 = MyProcess("jump")
    p1.start()    #start会自动调用run
    p1.join()
    print('主线程')

join 用法：（主进程等待join的子进程）

from multiprocessing import Process
import time
def fun(name):
    print("hi",name)

if __name__ == "__main__":
    li = []
    for i in range(5):
        pro=Process(target=fun,args=(i,))
        pro.start()
        li.append(pro)
    for i in li:
        i.join()
    print("hi,crazy")

执行结果：

1.4.2、守护进程

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

• 守护进程会在主进程代码执行结束后就终止
• 守护进程内无法再开启子进程,否则抛出异常：AssertionError: daemonic processes are not allowed to have children

from multiprocessing import Process
import time
def fun(name):
    print("hi",name)

if __name__ == "__main__":

    pro=Process(target=fun,args=("jump",))
    pro.daemon = True   #一定要放在start之前
    pro.start()
    print("hi,crazy")


#执行结果：hi，crazy  主进程结束整个程序结束

1.4.2、进程同步锁

多个任务可以同时在几个进程中并发处理，他们之间的运行没有顺序，一旦开启也不受我们控制。尽管并发编程让我们能更加充分的利用IO资源，但是也给我们带来了新的问题。当多个进程使用同一份数据资源的时候，就会引发数据安全或顺序混乱问题

import os
import time
from multiprocessing import Process,Lock

def work(lock,n):
    lock.acquire()
    print('%s: %s is running' % (n, os.getpid()))
    print(time.time())
    print('%s: %s is done' % (n, os.getpid()))
    lock.release()
if __name__ == '__main__':
    lock=Lock()
    for i in range(3):
        p=Process(target=work,args=(lock,i))
        p.start()

1.4.3、信号量（multiprocess.Semaphor）、事件（multiprocess.Event）

同步锁同时只允许一个进程更改数据，而信号量Semaphore是同时允许一定数量的进程更改数据

from multiprocessing import Process,Semaphore
import time,random

def go_eat(sem,user):
    sem.acquire()
    print('请%s用餐' %user)
    time.sleep(random.randint(0,3)) #模拟每个人的用餐时间不同
    sem.release()

if __name__ == '__main__':
    sem=Semaphore(5)
    li=[]
    for i in range(20):
        p=Process(target=go_eat,args=(sem,i,))
        p.start()
        li.append(p)
    for i in li:
        i.join()
    print("用餐结束")

Python进程中的事件用于主进程控制其他进程，事件主要提供了三个方法：set、clear、wait。事件处理的机制：全局定义了一个“Flag”，如果“Flag”值为 False，那么当程序执行 event.wait 方法执行时就会阻塞，如果“Flag”值为True，那么event.wait 方法执行时便不再阻塞。（clear：将“Flag”设置为False；set：将“Flag”设置为True）

from multiprocessing import Process, Event
import time, random
def car(e, n):
    while True:
        if not e.is_set():  # 进程刚开启，is_set()的值是Flase，模拟信号灯为红色
            print('红灯亮，car%s等着' % n)
            e.wait()    # 阻塞，等待is_set()的值变成True，模拟信号灯为绿色
            print('车%s 看见绿灯亮了' % n)
            time.sleep(random.randint(3, 6))
            if not e.is_set():  # 如果is_set()的值是Flase，也就是红灯，仍然回到while语句开始
                continue
            print('car%s开走了' %n)

def lights(e, inverval):
    while True:
        time.sleep(inverval)
        if e.is_set():

            e.clear()  # ---->将is_set()的值设置为False
            print('红灯亮了', e.is_set())
        else:
            e.set()    # ---->将is_set()的值设置为True
            print('绿灯亮了',e.is_set())


if __name__ == '__main__':
    e = Event()
    t = Process(target=lights, args=(e, 10))  # 创建一个进程控制红绿灯
    t.start()

    for i in range(10):
        p=Process(target=car,args=(e,i,))  # 创建10个进程控制10辆车
        p.start()

1.4.4、进程通讯

进程队列（multiprocessing.Queue），Queue是多进程安全的队列，可以使用Queue实现多进程之间的数据传递

方法：

q = multiprocessing.Queue([maxsize])
创建共享的进程队列。maxsize是队列中允许的最大项数。如果省略此参数，则无大小限制。底层队列使用管道和锁定实现。另外，还需要运行支持线程以便队列中的数据传输到底层管道中。 
Queue的实例q具有以下方法：

q.get( [ block [ ,timeout ] ] ) 
返回q中的一个项目。如果q为空，此方法将阻塞，直到队列中有项目可用为止。block用于控制阻塞行为，默认为True. 如果设置为False，将引发Queue.Empty异常（定义在Queue模块中）。timeout是可选超时时间，用在阻塞模式中。如果在制定的时间间隔内没有项目变为可用，将引发Queue.Empty异常。

q.get_nowait( ) 
同q.get(False)方法。

q.put(item [, block [,timeout ] ] ) 
将item放入队列。如果队列已满，此方法将阻塞至有空间可用为止。block控制阻塞行为，默认为True。如果设置为False，将引发Queue.Empty异常（定义在Queue库模块中）。timeout指定在阻塞模式中等待可用空间的时间长短。超时后将引发Queue.Full异常。

q.qsize() 
返回队列中目前项目的正确数量。此函数的结果并不可靠，因为在返回结果和在稍后程序中使用结果之间，队列中可能添加或删除了项目。在某些系统上，此方法可能引发NotImplementedError异常。


q.empty() 
如果调用此方法时 q为空，返回True。如果其他进程或线程正在往队列中添加项目，结果是不可靠的。也就是说，在返回和使用结果之间，队列中可能已经加入新的项目。

q.full() 
如果q已满，返回为True. 由于线程的存在，结果也可能是不可靠的（参考q.empty（）方法）

q.close() 
关闭队列，防止队列中加入更多数据。调用此方法时，后台线程将继续写入那些已入队列但尚未写入的数据，但将在此方法完成时马上关闭。如果q被垃圾收集，将自动调用此方法。关闭队列不会在队列使用者中生成任何类型的数据结束信号或异常。例如，如果某个使用者正被阻塞在get（）操作上，关闭生产者中的队列不会导致get（）方法返回错误。

q.cancel_join_thread() 
不会再进程退出时自动连接后台线程。这可以防止join_thread()方法阻塞。

q.join_thread() 
连接队列的后台线程。此方法用于在调用q.close()方法后，等待所有队列项被消耗。默认情况下，此方法由不是q的原始创建者的所有进程调用。调用q.cancel_join_thread()方法可以禁止这种行为。

from multiprocessing import Process,Queue

# 向queue中输入数据的函数
def inputQ(queue):
    info = time.ctime()
    queue.put(info)

# 向queue中输出数据的函数
def outputQ(queue):
    info = queue.get()
    print (info)
if __name__ == '__main__':
    record1 = []
    record2 = []
    queue = Queue()
    # 输入进程
    for i in range(10):
        process = Process(target=inputQ,args=(queue,))
        process.start()
        record1.append(process)

    # 输出进程
    for i in range(10):
        process = Process(target=outputQ,args=(queue,))
        process.start()
        record2.append(process)

    for p in record1:
        p.join()

    for p in record2:
        p.join()

执行结果：

基于队列的生产者、消费者模型

from multiprocessing import Process,Queue
import time,random,os
def consumer(que):
    while True:
        time.sleep(random.randint(1, 3))
        res=que.get()
        print('%s拿走了%s' %(os.getpid(),res))

def producer(q):
    while True:
        time.sleep(random.randint(1,3))
        res='100元'
        q.put(res)
        print('%s 生产了 %s' %(os.getpid(),res))

if __name__ == '__main__':
    q=Queue()
    #生产者们:即造钱者
    p1=Process(target=producer,args=(q,))

    #消费者们:即拿钱者
    c1=Process(target=consumer,args=(q,))

    #开始
    p1.start()
    c1.start()

进程管道（multiprocess.Pipe）

#创建管道的类：
Pipe([duplex]):在进程之间创建一条管道，并返回元组（conn1,conn2）,其中conn1，conn2表示管道两端的连接对象，强调一点：必须在产生Process对象之前产生管道
#参数介绍：
dumplex:默认管道是全双工的，如果将duplex射成False，conn1只能用于接收，conn2只能用于发送。
#主要方法：
    conn1.recv():接收conn2.send(obj)发送的对象。如果没有消息可接收，recv方法会一直阻塞。如果连接的另外一端已经关闭，那么recv方法会抛出EOFError。
    conn1.send(obj):通过连接发送对象。obj是与序列化兼容的任意对象
 #其他方法：
conn1.close():关闭连接。如果conn1被垃圾回收，将自动调用此方法
conn1.fileno():返回连接使用的整数文件描述符
conn1.poll([timeout]):如果连接上的数据可用，返回True。timeout指定等待的最长时限。如果省略此参数，方法将立即返回结果。如果将timeout射成None，操作将无限期地等待数据到达。
 
conn1.recv_bytes([maxlength]):接收c.send_bytes()方法发送的一条完整的字节消息。maxlength指定要接收的最大字节数。如果进入的消息，超过了这个最大值，将引发IOError异常，并且在连接上无法进行进一步读取。如果连接的另外一端已经关闭，再也不存在任何数据，将引发EOFError异常。
conn.send_bytes(buffer [, offset [, size]])：通过连接发送字节数据缓冲区，buffer是支持缓冲区接口的任意对象，offset是缓冲区中的字节偏移量，而size是要发送字节数。结果数据以单条消息的形式发出，然后调用c.recv_bytes()函数进行接收    
 
conn1.recv_bytes_into(buffer [, offset]):接收一条完整的字节消息，并把它保存在buffer对象中，该对象支持可写入的缓冲区接口（即bytearray对象或类似的对象）。offset指定缓冲区中放置消息处的字节位移。返回值是收到的字节数。如果消息长度大于可用的缓冲区空间，将引发BufferTooShort异常。

from multiprocessing import Process,Pipe
import time,random,os
def consumer(que):
    while True:
        time.sleep(random.randint(1, 3))
        r,l=que
        print('%s拿走了%s' %(os.getpid(),l.recv()))   #接收

def producer(q):
    while True:
        time.sleep(random.randint(1,3))
        res='100元'
        r,l = q
        r.send(res)  #发送
        print('%s 生产了 %s' %(os.getpid(),res))

if __name__ == '__main__':
    l,r=Pipe()
    #生产者们:即造钱者
    p1=Process(target=producer,args=((l,r),))

    #消费者们:即拿钱者
    c1=Process(target=consumer,args=((l,r),))

    #开始
    p1.start()
    c1.start()

ps：在生产者中关闭管道不会有任何效果，除非消费者也关闭了相同的管道端点（队列就是管道加了锁）

1.4.5、进程之间数据共享（multiprocess.Manager）

进程间数据是独立的，可以借助于队列或管道实现通信，二者都是基于消息传递的。虽然进程间数据独立，但可以通过Manager实现数据共享

from multiprocessing import Manager,Process,Lock
def fun(d,lock):
    with lock: #不加锁而操作共享的数据,肯定会出现数据错乱
        d['count']-=1

if __name__ == '__main__':
    lock=Lock()
    with Manager() as dicts:
        dic=dicts.dict({'count':100})  #封装数据
        li=[]
        for i in range(100):
            p=Process(target=fun,args=(dic,lock))
            li.append(p)
            p.start()
        for p in li:
            p.join()
        print(dic)

1.4.6、进程池（multiprocess.Pool）

 Pool([numprocess  [,initializer [, initargs]]]):创建进程池 
参数介绍：
numprocess:要创建的进程数，如果省略，将默认使用cpu_count()的值
initializer：是每个工作进程启动时要执行的可调用对象，默认为None
initargs：是要传给initializer的参数组

主要方法：
 p.apply(func [, args [, kwargs]]):在一个池工作进程中执行func(*args,**kwargs),然后返回结果。需要强调的是：此操作并不会在所有池工作进程中并执行func函数。如果要通过不同参数并发地执行func函数，必须从不同线程调用p.apply()函数或者使用p.apply_async()

p.apply_async(func [, args [, kwargs]]):在一个池工作进程中执行func(*args,**kwargs),然后返回结果。此方法的结果是AsyncResult类的实例，callback是可调用对象，接收输入参数。当func的结果变为可用时，将理解传递给callback。callback禁止执行任何阻塞操作，否则将接收其他异步操作中的结果。
 
p.close():关闭进程池，防止进一步操作。如果所有操作持续挂起，它们将在工作进程终止前完成
p.jion():等待所有工作进程退出。此方法只能在close（）或teminate()之后调用

 使用：

#异步：
from multiprocessing import Pool
import os,time
def fun(n):
    print('%s run' %os.getpid())
    return n**2

if __name__ == '__main__':
    p=Pool(4) 
    #进程池中从无到有创建4个进程,以后一直是这4个进程执行任务直到结束
    li=[]
    for i in range(10):
        res=p.apply_async(fun,args=(i,)) 
        li.append(res)

    #异步apply_async用法：如果使用异步提交的任务，主进程需要使用jion，等待进程池内任务都处理完，然后可以用get收集结果，否则主进程结束，进程池可能还没来得及执行，也就跟着一起结束了
    p.close()
    p.join()
    for res in li:
        print(res.get()) #使用get来获取apply_aync的结果,如果是apply,则没有get方法,因为apply是同步执行,立刻获取结果,也根本无需get

#同步
from multiprocessing import Pool
import os,time
def fun(n):
    print('%s run' %os.getpid())
    return n**2

if __name__ == '__main__':
    p=Pool(4)
    #进程池中从无到有创建4个进程,以后一直是这三个进程在执行任务
    li=[]
    for i in range(10):
        res=p.apply(fun,args=(i,))
        #同步调用，直到本次任务执行完毕拿到res，同步调用都会在原地等着
        li.append(res)
    print(li)

需要回调函数的场景：进程池中任何一个任务一旦处理完了，就立即告知主进程，我处理好了，你可以处理我的结果了。主进程则调用一个函数去处理该结果，该函数即回调函数

from multiprocessing import Pool
import requests,json,os

def get(url):
    print('<进程%s> get %s' %(os.getpid(),url))
    respone=requests.get(url)
    if respone.status_code == 200:
        return {'url':url,'text':respone.text}
def pasrse(res):
    print('<进程%s> callback %s' %(os.getpid(),res['url']))
    parse_res='url:<%s> size:[%s]
' %(res['url'],len(res['text']))
    with open('db.txt','a') as f:
        f.write(parse_res)


if __name__ == '__main__':
    urls=[
        'https://www.baidu.com',
        'https://www.python.org',
        'https://www.163.com',
        'https://www.github.com/',
        'http://www.sina.com.cn/'
    ]

    p=Pool(2)
    res_li=[]
    for url in urls:
        res=p.apply_async(get,args=(url,),callback=pasrse)  #回调函数
        res_li.append(res)

    p.close()
    p.join()
    print([res.get() for res in res_li])

　　

二、线程

2.1、什么是线程

线程，有时被称为轻量进程，是程序执行流的最小单元。进程只是用来把资源集中到一起（进程只是一个资源单位，或者说资源集合），而线程才是cpu上的执行单位。

多线程（即多个控制线程）的概念就是在一个进程中存在多个控制线程，多个控制线程共享该进程的地址空间，相当于一个车间内有多条流水线，都共用一个车间的资源。

2.2、线程与进程的区别

• 地址空间和其它资源（如打开文件）：进程间相互独立，同一进程的各线程间共享。某进程内的线程在其它进程不可见
• 通信：进程间通信，线程间可以直接读写进程数据段（如全局变量）来进行通信——需要进程同步和互斥手段的辅助，以保证数据的一致性
• 调度和切换：线程上下文切换比进程上下文切换要快得多
• 在多线程操作系统中，进程不是一个可执行的实体

2.3、为何要用多线程

• 多线程共享一个进程的地址空间
• 线程比进程更轻量级，线程比进程更容易创建、撤销，在许多操作系统中创建一个线程比创建一个进程要快10-100倍，在有大量线程需要动态和快速修改时，这一特性很有用
• 若多个线程都是cpu密集型的，那么并不能获得性能上的增强，但是如果存在大量的计算和大量的I/O处理，拥有多个线程允许这些活动彼此重叠运行，从而会加快程序执行的速度。
• 在多cpu系统中，为了最大限度的利用多核，可以开启多个线程，比开进程开销要小的多。（这一条并不适用于python）

2.4、Python中的线程

2.4.1、全局解释器锁（GIL）

Python代码的执行由Python解释器来控制。Python在设计之初就考虑到要在主循环中，同时只有一个线程在执行。虽然 Python 解释器中可以“运行”多个线程，但在任意时刻只有一个线程在解释器中运行。对Python解释器的访问由全局解释器锁(GIL)来控制，正是这个锁能保证同一时刻只有一个线程在运行。

　　在多线程环境中，Python 解释器执行分析：

分析：
#1.100个线程去抢GIL锁，即抢执行权限
#2. 肯定有一个线程先抢到GIL（称线程1），然后开始执行，一旦执行就会拿lock.acquire()
#3. 极有可能线程1还未运行完毕，就有另外一个线程2抢到GIL，然后开始运行，但线程2发
#现互斥锁lock还未被线程1释放，于是阻塞，被迫交出执行权限，即释放GIL
#4.直到线程1重新抢到GIL，开始从上次暂停的位置继续执行，直到正常释放互斥锁lock，然后其他的线程再重复2 3 4的过程

2.4.2、threading模块

创建进程：

#方式1：
from threading import Thread
import time
def say(name):
    time.sleep(2)
    print('%s say hello' %name)

if __name__ == '__main__':
    t=Thread(target=say,args=('crazy',))
    t.start()
    print('主线程运行结束')

#方式2
from threading import Thread
import time
class Say(Thread):
    def __init__(self,name):
        super().__init__()
        self.name=name
    def run(self):
        time.sleep(2)
        print('%s say hello' % self.name)


if __name__ == '__main__':
    t = Say('jump')
    t.start()  #执行run
    print('主线程运行结束')

其他方法：

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

主进程等待子进程结束（join）

from threading import Thread
import time
def say(name):
    time.sleep(2)
    print('%s say hello' %name)

if __name__ == '__main__':
    t=Thread(target=say,args=('crazy',))
    t.start()
    t.join()
    print('主线程运行结束')

守护线程

from threading import Thread
import time
def say(name):
    time.sleep(2)
    print('%s say hello' %name)

if __name__ == '__main__':
    t=Thread(target=say,args=('crazy',))
    t.setDaemon(True)
    t.start()

    print('主线程运行结束')

#返回“主线程运行结束”  t作为主进程的守护进程，主进程结束它也跟着结束

#主进程在其代码结束后就已经算运行完毕了（守护进程在此时就被回收）,然后主进程会一直等非守护的子进程都运行完毕后回收子进程的资源(否则会产生僵尸进程)，才会结束，

#主线程在其他非守护线程运行完毕后才算运行完毕（守护线程在此时就被回收）。因为主线程的结束意味着进程的结束，进程整体的资源都将被回收，而进程必须保证非守护线程都运行完毕后才能结束。

同步锁与递归锁

同步通常被用来实现对共享资源的同步访问。为每一个共享资源创建一个Lock对象，当你需要访问该资源时，调用acquire方法来获取锁对象（如果其它线程已经获得了该锁，则当前线程需等待其被释放），待资源访问完后，再调用release方法释放锁

#不加锁，100个线程做100的递减
from threading import Thread
import time
def fun():
    global n
    temp=n
    time.sleep(0.1)
    n=temp-1
if __name__ == '__main__':
    n=100
    l=[]
    for i in range(100):
        p=Thread(target=fun)
        l.append(p)
        p.start()
    for p in l:
        p.join()

    print(n) #结果可能为99、98

#加锁
from threading import Thread,Lock
import time
def fun(lock):
    lock.acquire()
    global n
    temp=n
    time.sleep(0.1)
    n=temp-1
    lock.release()
if __name__ == '__main__':
    n=100
    l=[]
    lock=Lock()
    for i in range(100):
        p=Thread(target=fun,args=(lock,))
        l.append(p)
        p.start()
    for p in l:
        p.join()

    print(n) #结果为0

死锁： 是指两个或两个以上的进程或线程在执行过程中，因争夺资源而造成的一种互相等待的现象，若无外力作用，它们都将无法推进下去。此时称系统处于死锁状态或系统产生了死锁，这些永远在互相等待的进程称为死锁进程。如下：

from threading import Thread,Lock
import time
mutexA=Lock()
mutexB=Lock()

class MyThread(Thread):
    def run(self):
        self.func1()
        self.func2()
    def func1(self):
        mutexA.acquire()
        print('%s 拿到A锁' %self.name)

        mutexB.acquire()
        print('%s 拿到B锁' %self.name)
        mutexB.release()

        mutexA.release()

    def func2(self):
        mutexB.acquire()
        print('%s 拿到B锁' %self.name)
        time.sleep(2)

        mutexA.acquire()
        print('%s 拿到A锁' %self.name)
        mutexA.release()

        mutexB.release()

if __name__ == '__main__':
    for i in range(10):
        t=MyThread()
        t.start()


'''
Thread-1 拿到A锁
Thread-1 拿到B锁
Thread-1 拿到B锁
Thread-2 拿到A锁
然后就卡住，死锁了
'''

解决方法，递归锁，在Python中为了支持在同一线程中多次请求同一资源，python提供了可重入锁RLock。这个RLock内部维护着一个Lock和一个counter变量，counter记录了acquire的次数，从而使得资源可以被多次require。直到一个线程所有的acquire都被release，其他的线程才能获得资源。上面的例子如果使用RLock代替Lock，则不会发生死锁

from threading import Thread,RLock
import time
class MyThread(Thread):
    def actionA(self):
        r_lcok.acquire() #count=1
        print(self.name,"A锁",)
        time.sleep(2)
        r_lcok.acquire() #count=2
        print(self.name, "B锁",)
        time.sleep(1)
        r_lcok.release() #count=1
        r_lcok.release() #count=0
    def actionB(self):
        r_lcok.acquire()
        print(self.name, "B锁")
        time.sleep(2)
        r_lcok.acquire()
        print(self.name, "A锁")
        time.sleep(1)
        r_lcok.release()
        r_lcok.release()
    def run(self):
        self.actionA()
        self.actionB()
if __name__ == '__main__':

    # A=threading.Lock()
    # B=threading.Lock()

    r_lcok=RLock()

    for i in range(5):
        t=MyThread()
        t.start()

定时器：（Timer）

from threading import Timer


def hello(y):
    print("hello",y)


t = Timer(1, hello,args=("jump",))  
t.start()

#1s后输出hello,jump

信号量：（Semaphore）

同进程的用法一样，实例如下：

from threading import Thread,Semaphore
import threading
import time
def func():
    sm.acquire()
    print('%s name' %threading.current_thread().getName())
    time.sleep(2)
    sm.release()
if __name__ == '__main__':
    sm=Semaphore(5)  #每次5个
    for i in range(17):
        t=Thread(target=func)
        t.start()

事件：（Event）

同进程用法一样

event.isSet()：返回event的状态值；
event.wait()：如果 event.isSet()==False将阻塞线程；
event.set()： 设置event的状态值为True，所有阻塞池的线程激活进入就绪状态，等待操作系统调度； 
event.clear()：恢复event的状态值为False

实例：

import threading,time
class Boss(threading.Thread):
    def run(self):
        print("BOSS：今晚大家都要加班到22:00。")
        print(event.isSet())# False
        event.set()
        time.sleep(5)
        print("BOSS：<22:00>可以下班了。")
        print(event.isSet())
        event.set()

class Worker(threading.Thread):
    def __init__(self,name):
        super().__init__()
        self.wname=name
    def run(self):
        event.wait()#    一旦event被设定，等同于pass
        print("Worker%s：命苦啊！"%self.wname)
        time.sleep(1)
        event.clear()
        event.wait()
        print("Worker%s:OhYeah!"%self.wname)

if __name__=="__main__":
    event=threading.Event()
    threads=[]
    for i in range(5):
        threads.append(Worker(i))
    threads.append(Boss())
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    print("ending.....")

线程队列：（queue）

使用import queue，用法与进程Queue一样

方法：

import queue  （三种模式，先进先出、优先级、先进后出，默认先进先出）
#先进先出
q=queue.Queue()   #创建对象
#先进后出
q=queue.LifoQueue (先进后出)
#优先级
q=queue.PriorityQueue()  
q.put(2,"xxx")  2为优先级
#put进入一个元组,元组的第一个元素是优先级(通常是数字,也可以是非数字之间的比较),数字越小优先级越高

q.put("xxx") 增加值
q.get() 取值

q.qsize()  队列里面的数量
q.empty()   是否为空
q.full()     是否为满
q.task_done() 发送信号
q.join() 等待信号

　　

三、协程

3.1、什么是协程

协程：是单线程下的并发，又称微线程，纤程。英文名Coroutine。一句话说明什么是线程：协程是一种用户态的轻量级线程，即协程是由用户程序自己控制调度的

优点：

• 协程的切换开销更小，属于程序级别的切换，操作系统完全感知不到，因而更加轻量级
• 单线程内就可以实现并发的效果，最大限度地利用cpu

缺点：

• 协程的本质是单线程下，无法利用多核，可以是一个程序开启多个进程，每个进程内开启多个线程，每个线程内开启协程
• 协程指的是单个线程，因而一旦协程出现阻塞，将会阻塞整个线程

总结：

• 必须在一个单线程里实现并发
• 修改共享数据不需加锁
• 用户程序里自己保存多个控制流的上下文栈
• 附加：一个协程遇到IO操作自动切换到其它协程（如何实现检测IO，yield、greenlet都无法实现，就用到了gevent模块（select机制））

3.2、Greenlet模块

安装：pip3 install greenlet

#例子
from greenlet import greenlet
def say(name):
    print('%s say 1' %name)
    g2.switch(name)
    print('%s say 2' %name)
    g2.switch()
def jump(name):
    print('%s jump 1' %name)
    g1.switch()
    print('%s jump 2' %name)
g1=greenlet(say)
g2=greenlet(jump)

g1.switch('crazy')

执行结果：

单纯的切换（在没有io的情况下或者没有重复开辟内存空间的操作），反而会降低程序的执行速度，greenlet只是提供了一种更加便捷的切换方式，当切到一个任务执行时如果遇到io，那就原地阻塞，仍然是没有解决遇到IO自动切换来提升效率的问题

import time
def fun1():
    res=1
    for i in range(10000000):
        res+=i
def fun2():
    res=1
    for i in range(10000000):
        res*=i
start=time.time()
fun1()
fun2()
stop=time.time()
print("rtime:",stop-start)

from greenlet import greenlet
def f1():
    res=1
    for i in range(10000000):
        res+=i
        g2.switch()

def f2():
    res=1
    for i in range(10000000):
        res*=i
        g1.switch()

start=time.time()
g1=greenlet(f1)
g2=greenlet(f2)
g1.switch()
stop=time.time()
print('gtime:',stop-start)

结果：

3.3、Gevent模块

 安装：pip3 install gevent

Gevent 是一个第三方库，可以轻松通过gevent实现并发同步或异步编程，在gevent中用到的主要模式是Greenlet, 它是以C扩展模块形式接入Python的轻量级协程。 Greenlet全部运行在主程序操作系统进程的内部，但它们被协作式地调度（遇到IO阻塞时会自动切换任务）

import time
def fun1():
    res=1
    for i in range(10000000):
        res+=i
def fun2():
    res=1
    for i in range(10000000):
        res*=i
start=time.time()
fun1()
fun2()
stop=time.time()
print("rtime:",stop-start)

import gevent
def fun1():
    res=1
    for i in range(10000000):
        res+=i
def fun2():
    res=1
    for i in range(10000000):
        res*=i
st=time.time()
g1=gevent.spawn(fun1)
g2=gevent.spawn(fun2)
gevent.joinall([g1,g2])
sp=time.time()
print("gtime",sp-st)

执行结果：

或者看下面这个例子：

from gevent import monkey;monkey.patch_all()
import gevent
import time
def say():
    print('say 1')
    time.sleep(2)
    print('say 2')

def jump():
    print('jump 1')
    time.sleep(1)
    print('jump 2')

g1=gevent.spawn(say)
g2=gevent.spawn(jump)
gevent.joinall([g1,g2])
print('over')

　结果：

ps：需要将from gevent import monkey;monkey.patch_all()放到文件的开头，（用于识别IO的猴子补丁）

gevent并发socket：

#协程socket并发server端
from gevent import monkey;monkey.patch_all()
from socket import *
import gevent
def server(server_ip,port):
    s=socket(AF_INET,SOCK_STREAM)
    s.bind((server_ip,port))
    s.listen(10)
    while True:
        conn,addr=s.accept()
        gevent.spawn(talk,conn,addr)
def talk(conn,addr):
    try:
        while True:
            res=conn.recv(1024)
            print(addr)
            print('client %s:%s msg: %s' %(addr[0],addr[1],res))
            conn.send(res.upper())
    except Exception as e:
        print(e)
    finally:
        conn.close()
if __name__ == '__main__':
    server('127.0.0.1',8080)

#客户端
from socket import *
client=socket(AF_INET,SOCK_STREAM)
client.connect(('127.0.0.1',8080))
while True:
    msg=input('>>: ').strip()
    if not msg:continue
    client.send(msg.encode('utf-8'))
    msg=client.recv(1024)
    print(msg.decode('utf-8'))

 gevent爬虫：

#爬虫
from gevent import monkey;monkey.patch_all()
import gevent
import requests
import time

def get_page(url):
    response=requests.get(url)
    if response.status_code == 200:
        with open("1.txt","a",encoding="utf-8") as f:
            f.write(response.text)
        print('GET: %s,size:%s' % (url,len(response.text)))

sttime=time.time()
gevent.joinall([
                gevent.spawn(get_page,'https://www.yahoo.com/'),
                gevent.spawn(get_page,'https://www.python.org/'),
                gevent.spawn(get_page,'https://github.com')
                ])
sptime=time.time()
print("gtime",sptime-sttime)

urls=['https://www.python.org','https://www.yahoo.com','https://github.com']
stime=time.time()
for url in urls:
    get_page(url)
ptime=time.time()
print("ptime",ptime-stime)

执行结果：

四、IO多路复用

4.1、IO模型介绍（五种IO Model）：

    * blocking IO           阻塞IO
    * nonblocking IO      非阻塞IO
    * IO multiplexing      IO多路复用
    * signal driven IO     信号驱动IO
    * asynchronous IO    异步IO
    由signal driven IO（信号驱动IO）在实际中并不常用，所以主要介绍其余四种IO Model

阻塞IO

最简单的IO模型，用户线程在读写时被阻塞。数据拷贝指请求到的数据先存放在内核空间, 然后从内核空间拷贝至程序的缓冲区

用户线程在IO过程中被阻塞，不能做任何事情，对CPU的资源利用率不高

非阻塞IO

用户线程不断发起IO请求. 数据未到达时系统返回一状态值; 数据到达后才真正读取数据。用户线程每次请求IO都可以立即返回，但是为了拿到数据，需不断轮询，无谓地消耗了大量的CPU。一般很少直接使用这种模型，而是在其他IO模型中使用非阻塞IO这一特性

IO多路复用（事件驱动IO）

IO多路复用建立在内核提供的阻塞函数select上，用户先将需要进行IO操作的socket添加到select中，然后等待阻塞函数select返回。当数据到达后，socket被激活，select返回，用户线程就能接着发起read请求。

看起来和加了循环的同步阻塞IO差不多?实际上, 我们可以给select注册多个socket, 然后不断调用select读取被激活的socket，实现在同一线程内同时处理多个IO请求的效果。

• 如果处理的连接数不是很高的话，使用select/epoll的web server不一定比使用multi-threading + blocking IO的web server性能更好，可能延迟还更大。select/epoll的优势并不是对于单个连接能处理得更快，而是在于能处理更多的连接。
•  在多路复用模型中，对于每一个socket，一般都设置成为non-blocking，但是整个用户的process其实是一直被block的。只不过process是被select这个函数block，而不是被socket IO给block

当用户进程调用了select，那么整个进程会被block，而同时，kernel会“监视”所有select负责的socket，当任何一个socket中的数据准备好了，select就会返回。这个时候用户进程再调用read操作，将数据从kernel拷贝到用户进程

ps：

• select函数返回结果中如果有文件可读了，那么进程就可以通过调用accept()或recv()来让kernel将位于内核中准备到的数据copy到用户区
•  select的优势在于可以处理多个连接，不适用于单个连接

异步IO

真正的异步IO需要操作系统更强的支持。
IO多路复用模型中，数据到达内核后通知用户线程，用户线程负责从内核空间拷贝数据;
而在异步IO模型中，当用户线程收到通知时，数据已经被操作系统从内核拷贝到用户指定的缓冲区内，用户线程直接使用即可

（用户进程发起read操作之后，立刻就可以开始去做其它的事。而另一方面，从kernel的角度，当它受到一个asynchronous read之后，首先它会立刻返回，所以不会对用户进程产生任何block。然后，kernel会等待数据准备完成，然后将数据拷贝到用户内存，当这一切都完成之后，kernel会给用户进程发送一个signal，告诉它read操作完成了）

 4.2、select、poll、epoll IO多路复用介绍

• select
  select最早于1983年出现在4.2BSD中，它通过一个select()系统调用来监视多个文件描述符的数组，当select()返回后，该数组中就绪的文件描述符便会被内核修改标志位，使得进程可以获得这些文件描述符从而进行后续的读写操作。 
  select目前几乎在所有的平台上支持 
  select的一个缺点在于单个进程能够监视的文件描述符的数量存在最大限制，在Linux上一般为1024，不过可以通过修改宏定义甚至重新编译内核的方式提升这一限制。 
  另外，select()所维护的存储大量文件描述符的数据结构，随着文件描述符数量的增大，其复制的开销也线性增长。同时，由于网络响应时间的延迟使得大量TCP连接处于非活跃状态，但调用select()会对所有socket进行一次线性扫描，所以这也浪费了一定的开销。

• poll 
  它和select在本质上没有多大差别，但是poll没有最大文件描述符数量的限制。 
  一般也不用它，相当于过渡阶段

• epoll 
  直到Linux2.6才出现了由内核直接支持的实现方法，那就是epoll。被公认为Linux2.6下性能最好的多路I/O就绪通知方法。windows不支持 
  没有最大文件描述符数量的限制。 
  比如100个连接，有两个活跃了，epoll会告诉用户这两个两个活跃了，直接取就ok了，而select是循环一遍
  （了解）epoll可以同时支持水平触发和边缘触发（Edge Triggered，只告诉进程哪些文件描述符刚刚变为就绪状态，它只说一遍，如果我们没有采取行动，那么它将不会再次告知，这种方式称为边缘触发），理论上边缘触发的性能要更高一些，但是代码实现相当复杂。 
  另一个本质的改进在于epoll采用基于事件的就绪通知方式。在select/poll中，进程只有在调用一定的方法后，内核才对所有监视的文件描述符进行扫描，而epoll事先通过epoll_ctl()来注册一个文件描述符，一旦基于某个文件描述符就绪时，内核会采用类似callback的回调机制，迅速激活这个文件描述符，当进程调用epoll_wait()时便得到通知。 
  所以市面上上见到的所谓的异步IO，比如nginx、Tornado、等，我们叫它异步IO，实际上是IO多路复用。

#select server端
import socket
import select
sk=socket.socket()
sk.bind(("127.0.0.1",8080))
sk.listen(5)
li=[sk,]
while True:
    r,w,e=select.select(li,[],[],5)  #r为输入，w为输出，e为错误，监视5s
    for obj in r:
        if obj==sk:
            conn,add=obj.accept()
            print(conn)
            li.append(conn)
        else:
            data_byte=obj.recv(1024)
            print(str(data_byte,'utf8'))
            inp=input('回答%s号客户>>>'%li.index(obj))
            obj.sendall(bytes(inp,'utf8'))

    print('>>',r)

#客户端
import socket
sk = socket.socket()
sk.connect(('127.0.0.1', 8080))
while True:
    inp = input(">>>>")
    sk.sendall(bytes(inp, "utf8"))
    data = sk.recv(1024)
    print(str(data, 'utf8'))

4.3、selectors模块

这三种IO多路复用模型在不同的平台有着不同的支持，而epoll在windows下就不支持，好在我们有selectors模块，帮我们默认选择当前平台下最合适的

#selectors模块server端
import selectors
import socket
sel = selectors.DefaultSelector()  #创建对象
def accept(sock, mask):
    conn, addr = sock.accept()
    print('accepted', conn, 'from', addr)
    conn.setblocking(False)
    sel.register(conn, selectors.EVENT_READ, read)

def read(conn, mask):
    try:
        data = conn.recv(1024)
        if not data:
            raise Exception
        print('echoing', repr(data), 'to', conn)
        conn.send(data)
    except Exception as e:
        print('closing', conn)
        sel.unregister(conn)
        conn.close()
sock = socket.socket()
sock.bind(('localhost', 8080))
sock.listen(5)
sock.setblocking(False)

sel.register(sock, selectors.EVENT_READ, accept)  #注册

while True:
    events = sel.select()   #在此循环接收
    for key, mask in events:
        print("key",key)
        callback = key.data
        callback(key.fileobj, mask)

#client端
import socket
sk = socket.socket()
sk.connect(('127.0.0.1', 8080))
while True:
    inp = input(">>>>")
    sk.sendall(bytes(inp, "utf8"))
    data = sk.recv(1024)
    print(str(data, 'utf8'))