Python并发编程之协程

协程

一、协程的本质：

单线程实现并发,在应用程序里控制多个任务的切换+保存状态

二、协程的目的：

• 想要在单线程下实现并发
• 并发指的是多个任务看起来是同时运行的
• 并发=切换+保存状态

三、补充：

• yiled可以保存状态，yield的状态保存与操作系统的保存线程状态很像，但是yield是代码级别控制的，更轻量级
• send可以把一个函数的结果传给另外一个函数，以此实现单线程内程序之间的切换
• 如何实现检测IO，yield、greenlet都无法实现，就用到了gevent模块（select机制）

四、优点

• 应用程序级别速度要远远高于操作系统的切换

五、缺点

• 多个任务一旦有一个阻塞没有切，整个线程都阻塞在原地，该线程内的其他的任务都不能执行了
• 一旦引入协程，就需要检测单线程下所有的IO行为,实现遇到IO就切换,少一个都不行，因为如果一个任务阻塞了，整个线程就阻塞了，其他的任务即便是可以计算，但是也无法运行了

注意：单纯地切换反而会降低运行效率

#并发执行
import time
 
def producer():
    g=consumer()
    next(g)
    for i in range(100):
        g.send(i)
 
def consumer():
    while True:
        res=yield
 
start_time=time.time()
producer()
stop_time=time.time()
print(stop_time-start_time)
 
#串行
import time
 
def producer():
    res=[]
    for i in range(10000000):
        res.append(i)
    return res
 
 
def consumer(res):
    pass
 
start_time=time.time()
res=producer()
consumer(res)
stop_time=time.time()
print(stop_time-start_time)

greenlet

greenlet只是提供了一种比generator更加便捷的切换方式，当切到一个任务执行时如果遇到io，那就原地阻塞，仍然是没有解决遇到IO自动切换来提升效率的问题。

注意：单纯的切换（在没有io的情况下或者没有重复开辟内存空间的操作），反而会降低程序的执行速度

#pip3 install greenlet
from greenlet import greenlet
import time
 
def eat(name):
    print('%s eat 1' %name)
    time.sleep(2)
    g2.switch('tom')
    print('%s eat 2' %name)
    g2.switch()
 
def play(name):
    print('%s play 1' %name )
    g1.switch()
    print('%s play 2' %name )
 
g1=greenlet(eat)
g2=greenlet(play)
 
g1.switch('tom')
 
"""
tom eat 1
tom play 1
tom eat 2
tom play 2
"""

gevent

遇到IO阻塞时会自动切换任务

一、用法：

• g1=gevent.spawn(func,1,,2,3,x=4,y=5)创建一个协程对象g1，spawn括号内第一个参数是函数名，如eat，后面可以有多个参数，可以是位置实参或关键字实参，都是传给函数eat的
• g2=gevent.spawn(func2)
• g1.join() #等待g1结束
• g2.join() #等待g2结束
• 或者上述两步合作一步：gevent.joinall([g1,g2])
• g1.value#拿到func1的返回值

二、补充：

• gevent.sleep(2)模拟的是gevent可以识别的io阻塞,
• 而time.sleep(2)或其他的阻塞,gevent是不能直接识别的需要用下面一行代码,打补丁,就可以识别了
• from gevent import monkey;monkey.patch_all()必须放到被打补丁者的前面，如time，socket模块之前或者我们干脆记忆成：要用gevent，需要将from gevent import monkey;monkey.patch_all()放到文件的开头

#pip3 install gevent
from gevent import monkey;monkey.patch_all()
import gevent
import time
 
def eat(name):
    print('%s eat 1' % name)
    time.sleep(3)
    print('%s eat 2' % name)
 
def play(name):
    print('%s play 1' % name)
    time.sleep(2)
    print('%s play 2' % name)
 
start_time=time.time()
g1=gevent.spawn(eat,'tom')
g2=gevent.spawn(play,'rose')
 
g1.join()
g2.join()
stop_time=time.time()
print(stop_time-start_time)
"""
tom eat 1
rose play 1
rose play 2
tom eat 2
3.003171920776367
"""
 
 
 
from gevent import monkey;monkey.patch_all()
import gevent
import time
 
def eat(name):
    print('%s eat 1' % name)
    time.sleep(3)
    print('%s eat 2' % name)
 
def play(name):
    print('%s play 1' % name)
    time.sleep(2)
    print('%s play 2' % name)
 
g1=gevent.spawn(eat,'tom')
g2=gevent.spawn(play,'rose')
 
# g1.join()
# g2.join()
gevent.joinall([g1,g2])

三、通过gevent实现单线程下的socket并发

from gevent import monkey;monkey.patch_all()一定要放到导入socket模块之前，否则gevent无法识别socket的阻塞

"""
服务端
#基于gevent实现
"""
from gevent import monkey,spawn;monkey.patch_all()
from socket import *
 
def communicate(conn):
    while True:
        try:
            data=conn.recv(1024)
            if not data:break
            conn.send(data.upper())
        except ConnectionResetError:
            break
 
    conn.close()
 
def server(ip,port):
    server = socket(AF_INET, SOCK_STREAM)
    server.bind((ip,port))
    server.listen(5)
 
    while True:
        conn, addr = server.accept()
        spawn(communicate,conn)
 
    server.close()
 
if __name__ == '__main__':
    g=spawn(server,'127.0.0.1',8090)
    g.join()
     
     
"""
客户端
"""
from socket import *
from threading import Thread,currentThread
 
def client():
    client=socket(AF_INET,SOCK_STREAM)
    client.connect(('127.0.0.1',8090))
 
    while True:
        client.send(('%s hello' %currentThread().getName()).encode('utf-8'))
        data=client.recv(1024)
        print(data.decode('utf-8'))
 
    client.close()
 
if __name__ == '__main__':
    for i in range(500):
        t=Thread(target=client)
        t.start()

asyncio

"""
网络编程
B/S和C/S架构
osi五层协议
应用层
传输层  端口/服务相关  四次路由器 四层交换机
网络层 ip 路由器 三层交换机
数据链路层 arp mac相关 网卡 交换机
物理层
socket/socketserver
tcp/udp
    tcp协议：面向连接的，可靠的，全双工的，流失传输协议
    三次握手
    四次挥手
    粘包现象
        在接受度粘 发送信息无边界，在接受端没有及时接受，在缓存端粘在一起
        在发送端粘，时间短，由于优化机制粘在一起的
    udp协议：面向数据报的，不可靠 无连接的协议
ip
arp

I/O操作 输入会让输出
    input输入到内存 input  read  recv  recvfrom   accept   connect  close
    output 从内存输出 print write  send   sendto   accept   connect  close

阻塞和非阻塞
阻塞： cpu不工作 accept recv connect等待
非阻塞：cpu工作


##############################

人机矛盾
    cpu利用率低
磁带存储+批处理
    降低数据的读取时间
    提高cpu的利用率
多道操作系统------在一个任务遇到io的时候主动让出cpu
    数据隔离
    时空复用
    能够在一个任务遇到io操作的时候主动把cpu让出来，给其他的任务使用
        切换：占时间，
        切换：操作系统做的
分时操作系统------给时间分片，让多个任务轮流使用cpu
    cpu的轮转
    每一个程序分配一个时间片
    切换：占时间
    反而降低来cpu的使用率
    提高了用户体验
分时操作系统+多道操作系统+实时操作系统
    多个程序一起在计算机中执行
    一个程序如果遇到io操作，切出去让出cpu
    一个程序没有遇到io，但是时间片到时了，切出去让出cpu
网络操作系统
分布式操作系统



##############################
进程：运行中的程序
程序和进程之间的区别
    程序只是一个文件
    进程是这个文件被cpu运行起来了
进程是计算机中最小的资源分配单位
每一个进程都有一个在操作系统中的唯一标识符：pid

操作系统调度进程的算法
    短作业优先
    先来先服务
    时间片轮转
    多级反馈算法（少在程序的一开始就使用io）

操作系统负责什么？
    调度进程先后执行的顺序，控制执行的时间等等
    资源的分配

并行与并发
    并行：两个程序两个cpu，每个程序分别占用一个cpu自己执行自己的
         看起来是同时执行，实际在每一个时间点上都在各自执行着
    并发：两个程序一个cpu每个程序交替在一个cpu上执行，看起来在同时执行，但是实际上仍然是串行
同步：
    调用这个函数，并等待这个函数结束
异步：
    调用一个函数，不等待这个方法结束，也不关心这个方法做了什么
阻塞：cpu不工作
非阻塞：cpu工作
同步阻塞：
    conn.recv
    调用函数的这个过程，有io操作
同步非阻塞：
    func() 没有io操作
    socket 非阻塞的tcp协议的时候
    调用函数（这个函数内部不存在io操作）
异步非阻塞：
    把func仍到其他任务里去执行里，没有io操作
    本身的任务和func任务各自执行各自的，没有io操作
异步阻塞：


双击图标发生的事情：
程序在开始运行之后，并不是立即开始执行代码，而是会进入就绪状态，等待操作系统调度开始运行
操作系统为其创建进程，分配一块空间，pid---》就绪（ready）---》运行（runing）---》阻塞（blocking）
运行（没有遇到io操作时间片到了）---》就绪
运行（遇到io）--》阻塞
运行--》运行完了--》结束进程
阻塞io结束---》就绪
阻塞影响了程序运行的效率

##############################
进程是计算机中最小的资源分配单位（进程负责圈资源）
    数据隔离的
    创建进程，时间开销大
    销毁进程，时间开销大
    进程之间切换，时间开销大
线程：是计算机中能被cpu调度的最小单位（线程是负责执行具体代码的）
    线程的创建，也需要一些开销（一个存储局部变量的结构，记录状态）
    创建，销毁，切换开销远远小于进程
    每个进程中至少有一个线程
    一个进程中的多个线程是可以共享这个进程的数据的

是进程中的一部分
每一个进程中至少有一个线程，线程是负责执行具体代码的
只负责执行代码，不负责存储共享的数据，也不负责资源分配


python中的线程比较特殊，所以进程也有可能被用到。

pid子进程
ppid父进程
在pycharm中启动的所有py程序都是pycharm的子进程


tcp: 可靠
https: 安全
单核 并发
多核 多个程序跑在多个cpu上，在同一时刻并行

##############################

操作系统
    1。计算机中所有的资源都是由操作系统分配的
    2。操作系统调度任务：时间分片，多道机制
    3。cpu的利用率是我们努力的指标

进程：开销大 数据隔离 资源分配单位 cpython下可以利用多核  由操作系统调度的
    进程的三状态：就绪 运行 阻塞
    multiprocessing模块
    Process-开启进程
    Lock-互斥锁
        为什么要在进程中加锁
        因为进程操作文件也会发生数据不安全
    Queue -队列 IPC机制（Pipe，redis，memcache，rabbitmq，kafka）
    Manager-提供数据共享机制

线程：开销小 数据共享 cpu调度单位 cpython下不能利用多核  由操作系统调度的
    GIL锁：全局解释器锁，Cpython解释器提供的，致了一个进程中多个线程同一时刻只有一个线程能访问CPU--多线程不能利用多核
    Thread类--能开启线程start，等待线程结束join
    Lock-互斥锁  互斥锁能在一个线程中连续acquire，效率相对高
    Rlock--递归锁 可以在一个线程中连续acquire，效率相对低
    死锁现象如何发生，如何避免？
    线程队列queue模块
        Queue
        LifoQueue
        PriorityQueue
池
    实例化一个池
    提交任务到池中，返回一个对象
        使用这个对象获取返回值
        回调函数
    阻塞等待池中的任务都结束

数据安全问题
数据隔离和通信


协程：多个任务在一条线程上来回切换，用户级别的，由我们自己写的python代码来控制切换的，是操作系统不可见的
在Cpython解释器下----协程和线程都不能利用多核，都是在一个CPU上轮流切换
    由于多线程本身就不能利用多核
    所以即便是开启来多个线程也只能轮流在一个cpu上执行
    协程如果把所有任务的IO操作都规避掉，只剩下需要使用CPU的操作
    就意味着协程就可以做到提高CPU利用率的效果

多线程和协程
    线程 切换需要操作系统，开销大 给操作系统的压力大
        操作系统对IO操作的感知更加灵敏
    协程 切换需要 python代码 开销小，用户操作可控  完全不会增加操作系统的压力
        用户级别能够对IO操作的感知比较低

协程：能够在一个线程下的多个任务之间来回切换，那么每一个任务都是一个协程
两种切换方式
    原生python完成  yield asyncio
    c语言完成的python模块 greenlet  gevent

1。一个线程中的阻塞都被其他的各种任务占满了
2。让操作系统觉得这个线程很忙
尽量的减少这个线程进入阻塞的状态
提高了单线程对CPU的利用率
3。多个任务在同一个线程中执行
也达到了一个并发的效果
规避了每一个任务的io操作
减少了线程的个数，减轻了操作系统的负担



我们写协程：在一条线程上最大限度的提高CPU的使用率
           在一个任务中遇到IO的时候就切换到其他任务
特点：
    开销很小，是用户级的，只能从用户级别感知的IO操作
    不能利用多核，数据共享，数据安全
模块和用法：
    gevent （扩展模块）基于greenlet（内置模块）切换   aiohttp爬虫模块基于asyncio    sanic异步框架
        先导入模块
        导入monkey，执行patch all
        写一个函数当作协程要执行的任务
        协程对象 = gevent.spawn(函数名，参数，)
        协程对象.join(),gevent.joinall([g1,g2...])

    分辨gevent是否识别了我们写的代码中的io操作的方法
        在patchall之前打印一下涉及到io操作的函数地址
        在patchall之后打印一下涉及到io操作的函数地址
        如果两个地址一致，说明gevent没有识别这个io，如果不一致说明识别了

    asyncio 基于yield机制切换的
    async 标识一个协程函数
    await 后面跟着一个asyncio模块提供的io操作的函数
    loop 事件循环，负责在多个任务之间进行切换的

    最简单的协程函数是如何完成的


"""






import os
import time
from multiprocessing import Process
from multiprocessing import Queue
def func(exp,q):
    print('start',os.getpid())
    ret = eval(exp)
    q.put(ret)
    time.sleep(1)
    print('end',os.getpid())

if __name__ == '__main__': #window下需要加上这一句
    q=Queue() #先进先出
    p=Process(target=func,args=('1+2+3',q)) #创建一个即将要执行func函数的进程对象
    # p.daemon=True #守护进程是随着主进程的代码的结束而结束的
    p.start() #异步非阻塞    调用开启进程的方法，但是并不等待这个进程真的开启
    # p.join() #同步阻塞直到p对应的进程结束之后才结束阻塞
    # print(p.is_alive())
    # p.terminate() # 异步非阻塞  强制结束一个子进程
    # print(p.is_alive()) #子进程还活着，因为操作系统还没来得及关闭进程
    # time.sleep(0.01)
    # print(p.is_alive())#操作系统已经响应了我们要关闭进程的需求，再去检测的时候，得到的结束是进程已经结束了
    print('main',os.getpid()) #主进程代码执行完毕，但是主进程没有结束，等待子进程结束
    print(q.get())

"""
操作系统创建进程的方式不同
Windows操作系统执行开启进程的代码
    实际上新的子进程需要通过import父进程的代码来完成数据的导入工作
    所以有一些内容我们只希望在父进程中完成，就写在if __name__ == '__main__': 下面

iOS和Linux是直接从内存级别去拷贝代码（fork），从父进程拷贝数据到子进程

主进程没有结束，等待子进程结束
主进程负责回收子进程的资源
如果子进程执行结束，父进程没有回收资源，那么这个子进程会变成一个僵尸进程    

主进程的结束逻辑：
    主进程的代码结束
    所有的子进程结束
    给子进程回收资源
    主进程结束
主进程怎么知道子进程结束的呢？ 监控文件
    基于网络，文件    
join方法
    把一个进程的结束事件封装成一个join方法
    执行join方法的效果就是阻塞直到这个子进程执行结束就结束阻塞
    
    在多个子进程中执行join

p=Process(target=函数名,args=(参数,))
进程对象和进程并没有直接的关系，只是存储来一些和进程相关的内容，此时此刻，操作系统还没有接到创建进程的指令 
p.start()开启来一个进程----这个方法相当于给来操作系统一条指令
satrt方法的非阻塞和异步的特点
    在执行开启进程这个方法的时候
    我们既不等待这个进程开启，也不等待操作系统给我们的相应
    这里只是负责通知操作系统去开启一个进程
    开启了一个子进程之后，主进程的代码和子进程的代码完全异步
    
p.daemon 守护进程是随着主进程代码的结束而结束的
所有的子进程都必须在主进程结束之前结束，由主进程来负责回收资源 

p.is_alive()
p.terminate() 强制结束一个子进程的


什么是异步非阻塞？
    terminate,start  
什么是同步阻塞
    join      
    
1.如果在一个并发的场景下，涉及到某部分内容是需要修改一些所有进程共享的数据资源，需要加锁来维护数据的安全
2.在数据安全的基础上，才考虑效率问题
3.同步存在的意义就是数据安全   

在主进程中实例化 lock=Lock()
在子进程中，对需要加锁的代码进行with lock:
    with lock: 相当于lock.acquire()和lock.release()
在进程中需要加锁的场景
    1.操作共享的数据资源(文件，数据库)
    2.对资源进行修改，删除操作    


进程之间的通讯--IPC（inter process communication） 
队列和管道都是IPC机制，队列进程之间数据安全，管道进程之间数据不安全
第三方工具（软件）提供给我们的IPC机制,集群的概念（高可用）
redis
memcache
kafka
rabbitmq
并发需求，高可用，断电保存数据，解耦


from multiprocessing import Queue
Queue（天生就是数据安全的） 基于 文件家族的socket，pickle和lock实现的
队列就是管道加锁
get()：阻塞的
get_nowait():非阻塞的

下面三个方法不推荐使用，多进程中不准确（）
q.empty() 判断队列是否为空
q.qsize() 返回队列中的数据个数
q.full() 判断队列是否为满



pipe 管道（不安全的）基于 文件家族的socket，pickle实现的
from multiprocessing import Pipe
pip=Pipe()
pip.send()
pip.recv()

解耦：修改  复用  可读性
把写在一起的大的功能分开成多个小的功能处理

joinablequeue
q.join() 阻塞  直到这个队列中所有的内容都被取走且task_done

multiprocessing中有一个manager类
封装了所有和进程相关的，数据共享，数据传递相关的数据类型
但是对于字典列表这一类的数据操作的时候会产生数据不安全
需要加锁解决问题，并且需要尽量少的使用这种方式

线程本身是可以利用多核的
cpython解释器中不能实现多线程利用多核 
垃圾回收机制 gc  引用计数+分代回收
专门有一条线程来完成垃圾回收
对每一个在程序中的变量统计引用计数

锁：GIL全局解释器锁
cpython解释器中的机制，导致了在同一个进程中多个线程不能同时利用多核。
python的多线程只能是并发不能是并行

线程即便有GIL，也会出现数据不安全的问题
1.操作的是全局变量
2.做以下操作：
    +=，-=，*=，/=先计算再赋值才容易出现数据不安全的问题
    包括list[0]+=1,dic['key']-=1



a=0
def func():
    global a
    a+=1

import dis
dis.dis(func)  #查看func的cpu指令

互斥锁是锁中的一种：在同一个线程中，不能连续acquire多次，（在一个线程中连续多次acquire会死锁）
死锁现象：有多把锁+多把锁交替使用，互斥锁在一个线程中连续acquire
递归锁 （在一个线程中连续多次acquire不会死锁）
好：在同一个进程中多次acquire也不会发生阻塞
不好：占用了更多资源

递归锁---将多把互斥锁变成了一把递归锁，递归锁也会发生死锁现象（多把递归锁交替使用的时候），以后的代码尽量用一把锁。
mutexB=mutexA=RLock()
注意不是：mutexB=RLock()  mutexA=RLock() 不然还是一样会发生死锁现象



保证了整个python程序中，只能有一个线程被CPU执行
原因;cpython解释器中特殊的垃圾回收机制
GIL锁导致来线程不能并行，但是可以并发
所以使用多线程并不影响高io型的操作
只会对高计算型的程序有效率上的影响
遇到高计算的：多进程+多线程  或者 分布式


遇到IO操作的时候
5亿条cpu指令/s
5-6cpu指令==一句python代码
几千万条python代码
web框架，几乎都是多线程


主线程什么时候结束？等待所有子线程结束之后才结束
主线程如果结束了，主进程也就结束了
t.ident  线程id
from threading import Thread,current_thread,active_count,enumerate
active_count()==len(enumerate())
current_thread.ident  线程id   在哪一个线程里，current_thread得到的就是这个当前线程的信息
在线程中不能从主线程结束一个子线程 没有terminate
不要在子线程里随便修改全局变量
守护线程一直等到所有的非守护线程都结束之后才结束
除了守护了主线程的代码之外也会守护子线程

p.daemon 守护进程是随着主进程代码的结束而结束的
非守护线程不结束，主线程也不结束，主线程结束了，主进程也结束
结束顺序：
非守护线程结束--》主线程结束--》主进程结束——》守护线程也结束



什么是生产者消费者模型
把一个产生数据并且处理数据的过程解耦
让生产数据的过程和处理数据的过程达到一个工作效率上的平衡
中间的容器，在多进程中使用队列或者可被join的队列，做到控制数据的量
当数据过剩的时候，队列的大小会控制消费者的行为
当数据严重不足的时候，队列会控制消费者的行为
并且还可以通过定期检测队列中元素的个数来调节生产者消费者的个数
爬虫：
    请求网页的平均时间是0.3s
    处理网页代码的时候是0.003s
    100倍，每启动100个线程生产数据，就启动一个线程来处理数据
web程序的server端
    每秒钟有6w条请求
    一个服务每s中只能处理2000条
    先写一个web程序，只负责一件事情，就是接收请求，然后把请求放到队列中（消息中间件）
    再写很多个server端，从队列中获取请求，然后处理，然后返回结果

# from queue import Queue #先进先出队列
import queue  #线程之间的通信，线程安全

先进先出：写一个server所有的用户的请求放在队列里，先来先服务的思想  q=queue.Queue(3)
后进先出：算法中    q=queue.LifoQueue(3) #后进先出->堆栈
优先级队列：自动的排序（vip号码段）告警级别    q=queue.PriorityQueue(3) #优先级队列   q.put((10,'one'))

池：预先的开启固定个数的进程数，当任务来临的时候，直接提交给已经开好的进程/线程，让其去执行就可以了
节省了进程，线程的开启 关闭 切换锁花费的时间
并且减轻了操作系统调度的负担

shutdown 关闭池之后就不能继续提交任务，并且会阻塞，直到已经提交的任务完成
result() 同步阻塞
submit() 异步非阻塞

5*20*500

是单独开启线程进程还是池？
如果只是开启一个子进程做一件事情，就可以单独开线程
有大量的任务等待程序去做，要达到一定的并发数，开启线程池
根据你程序的io操作也可以判定是用池还是不用池
socket大量的阻塞io ，socket server就没有用池，用的是线程
爬虫的时候，池

回调函数：执行完子线程任务之后直接调用对应的回调函数
        爬取网页 需要等待数据传输和网络上的响应高IO的--子线程
        分析网页，没有什么IO操作--这个操作没必要在子线程完成，交给回调函数
两件事情：存在IO操作的事情，基本不存在IO操作的事情
obj=tp.submit(io操作对应的函数)
obj.result() 是一个阻塞方法 获取返回值
obj.add_done_callback(计算型的事情)

obj=tp.submit(需要在子线程执行的函数名，参数)
obj.add_done_callback(子线程执行完毕之后要执行的代码对应的函数)

进程和线程都有锁
所有在线程中能工作的基本都不能在进程中工作
在进程中能够使用的基本在线程中也可以使用

没一个进程下面都有一个独立的垃圾回收机制的线程，不需要担心开了多进程以后，垃圾回收机制怎么处理


"""

from multiprocessing import process
from threading import Thread
import os
def tfunc():
    print(os.getpid())

def pfunc():
    print('pfunc-->',os.getpid())
    Thread(target=tfunc).start()

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




from multiprocessing import Queue
import queue
q=Queue(3)
q.put(1)
q.put(2)
q.put(3)
print('aaa')
# q.put(44444)  #当队列为满的时候再向队列中放数据，队列会阻塞
try:
    q.put_nowait(6) #当队列为满的时候再向队列中放数据，会报错并且会丢失数据
except queue.Full:
    pass
print('bbb')

print(q.get())
print(q.get())
print(q.get())
# print(q.get()) #在队列为空的时候会发生阻塞
try:
    q.get_nowait() #当队列为空的时候直接报错
except queue.Empty:
    pass





"""
##############################
协程：多个任务在一条线程上来回切换
我们写协程：在一条线程上最大限度的提高CPU的使用率
           在一个任务中遇到IO的时候就切换到其他任务
特点：
    开销很小，是用户级的，只能从用户级别感知的IO操作
    不能利用多核，数据共享，数据安全
模块和用法：
    gevent 基于greenlet切换
        先导入模块
        导入monkey，执行patch all
        写一个函数当作协程要执行的任务
        协程对象 = gevent.spawn(函数名，参数，)
        协程对象.join(),gevent.joinall([g1,g2...])

    分辨gevent是否识别了我们写的代码中的io操作的方法
        在patchall之前打印一下涉及到io操作的函数地址
        在patchall之后打印一下涉及到io操作的函数地址
        如果两个地址一致，说明gevent没有识别这个io，如果不一致说明识别了

    asyncio 基于yield机制切换的
    async 标识一个协程函数
    await 后面跟着一个asyncio模块提供的io操作的函数
    loop 事件循环，负责在多个任务之间进行切换的

    最简单的协程函数是如何完成的
"""



from gevent import monkey
monkey.patch_all()
import gevent
import time


def eat(name):
    print('%s eat 1' % name)
    time.sleep(3)
    print('%s eat 2' % name)


def play(name):
    print('%s play 1' % name)
    time.sleep(2)
    print('%s play 2' % name)
    return '返回值'

g_l=[]
for i in range(10):
    g = gevent.spawn(eat, 'tom')  #注意要遇到阻塞才会切换
    g_l.append(g)
gevent.joinall(g_l) #这就是阻塞的

g2=gevent.spawn(play,'rose')
g2.join()
g2.value #获取返回值,需要注意的是这个是一个属性，需要在join后才能获取到，不然就是None  此处打印 返回值


import asyncio
async def func(): #协程方法
    print("start")
    await asyncio.sleep(1)  #阻塞  需要一个关键字await告知是阻塞需要执行一个协程了，如果用await需要在函数之前加上async表示是一个协程函数
    print("end")
    return 123

async def func1():
    print("start")
    await asyncio.sleep(1)
    print("end")
    return 456

async def func2():
    print("start")
    await asyncio.sleep(1)
    print("end")
    return 789

#启动一个任务
loop=asyncio.get_event_loop() #创建一个事件循环
loop.run_until_complete(func()) #把func任务丢到事件循环中去执行    执行一个协程函数  和join差不多意思

#启动多个任务，并且没有返回值
loop=asyncio.get_event_loop()
wait_l=asyncio.wait([func(),func1(),func2()]) #执行多个协程函数
loop.run_until_complete(wait_l)


#启动多个任务，并且有返回值
loop=asyncio.get_event_loop()
f=loop.create_task(func())
f1=loop.create_task(func1())
f2=loop.create_task(func2())
task_l=[f,f1,f2] # 按照这个顺序取返回值
wait_l=asyncio.wait([f,f1,f2])
loop.run_until_complete(wait_l)
for i in task_l:
    print(i.result()) #返回值

#谁先回来先取谁的值
import asyncio
async def demo(i):
    print("start")
    await asyncio.sleep(10-i)
    print("end")
    return i,456

async def main():
    task_l=[]
    for i in range(10):
        task = asyncio.ensure_future(demo(i))
        task_l.append(task)

    for ret in asyncio.as_completed(task_l):
        res = await ret
        print(res)

loop=asyncio.get_event_loop()
loop.run_until_complete(main())

"""
await 阻塞 协程函数这里要切换出去，还能保证一会儿在切回来
await必须写在async函数里，async函数是协程函数
loop事件循环
所有的协程的执行，调度都离不开这个loop 
"""

a=0
def func():
    global a
    a+=1

import dis
dis.dis(func)  #查看func的cpu指令

"""
高性能异步爬虫
目的：在爬虫中使用异步实现高性能的数据爬取操作。

异步爬虫的方式：
    - 1.多线程，多进程（不建议）：
        好处：可以为相关阻塞的操作单独开启线程或者进程，阻塞操作就可以异步执行。
        弊端：无法无限制的开启多线程或者多进程。
    - 2.线程池、进程池（适当的使用）：
        好处：我们可以降低系统对进程或者线程创建和销毁的一个频率，从而很好的降低系统的开销。
        弊端：池中线程或进程的数量是有上限。

- 3.单线程+异步协程（推荐）：
    event_loop：事件循环，相当于一个无限循环，我们可以把一些函数注册到这个事件循环上，
    当满足某些条件的时候，函数就会被循环执行。

    coroutine：协程对象，我们可以将协程对象注册到事件循环中，它会被事件循环调用。
    我们可以使用 async 关键字来定义一个方法，这个方法在调用时不会立即被执行，而是返回
    一个协程对象。

    task：任务，它是对协程对象的进一步封装，包含了任务的各个状态。

    future：代表将来执行或还没有执行的任务，实际上和 task 没有本质区别。

    async 定义一个协程.

    await 用来挂起阻塞方法的执行。

"""
import asyncio

async def request(url):
    print('正在请求的url是',url)
    print('请求成功,',url)
    return url
#async修饰的函数，调用之后返回的一个协程对象
c = request('www.baidu.com')


#创建一个事件循环对象
loop = asyncio.get_event_loop()

#将协程对象注册到loop中，然后启动loop
loop.run_until_complete(c)



# #task的使用
# loop = asyncio.get_event_loop()
# #基于loop创建了一个task对象
# task = loop.create_task(c)
# print('task',task)
#
# loop.run_until_complete(task)



#future的使用
# loop = asyncio.get_event_loop()
# task = asyncio.ensure_future(c)
# print(task)
# loop.run_until_complete(task)
# print(task)

def callback_func(task):
    #result返回的就是任务对象中封装的协程对象对应函数的返回值
    print(task.result())

#绑定回调
loop = asyncio.get_event_loop()
task = asyncio.ensure_future(c)
#将回调函数绑定到任务对象中
task.add_done_callback(callback_func)
loop.run_until_complete(task)


#################多任务###############
import asyncio
import time

async def request(url):
    print('正在下载',url)
    #在异步协程中如果出现了同步模块相关的代码，那么就无法实现异步。
    # time.sleep(2)
    #当在asyncio中遇到阻塞操作必须进行手动挂起
    await asyncio.sleep(2)
    print('下载完毕',url)

start = time.time()
urls = [
    'www.baidu.com',
    'www.sogou.com',
    'www.goubanjia.com'
]

#任务列表：存放多个任务对象
stasks = []
for url in urls:
    c = request(url)
    task = asyncio.ensure_future(c)
    stasks.append(task)

loop = asyncio.get_event_loop()
#需要将任务列表封装到wait中
loop.run_until_complete(asyncio.wait(stasks))

print(time.time()-start)


##################################################
import requests
import asyncio
import time

start = time.time()
urls = [
    'http://127.0.0.1:5000/bobo',
    'http://127.0.0.1:5000/jay',
    'http://127.0.0.1:5000/tom'
]

async def get_page(url):
    print('正在下载',url)
    #requests.get是基于同步，必须使用基于异步的网络请求模块进行指定url的请求发送
    #aiohttp:基于异步网络请求的模块
    response = requests.get(url=url)
    print('下载完毕：',response.text)

tasks = []

for url in urls:
    c = get_page(url)
    task = asyncio.ensure_future(c)
    tasks.append(task)

loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(tasks))

end = time.time()

print('总耗时:',end-start)

#环境安装：pip install aiohttp
#使用该模块中的ClientSession
import requests
import asyncio
import time
import aiohttp

start = time.time()
# urls = [
#     'http://127.0.0.1:5000/bobo','http://127.0.0.1:5000/jay','http://127.0.0.1:5000/tom',
#     'http://127.0.0.1:5000/bobo', 'http://127.0.0.1:5000/jay', 'http://127.0.0.1:5000/tom',
#     'http://127.0.0.1:5000/bobo', 'http://127.0.0.1:5000/jay', 'http://127.0.0.1:5000/tom',
#     'http://127.0.0.1:5000/bobo', 'http://127.0.0.1:5000/jay', 'http://127.0.0.1:5000/tom',
#
# ]
from multiprocessing.dummy import Pool
pool = Pool(2)

urls = []
for i in range(10):
    urls.append('http://127.0.0.1:5000/bobo')
print(urls)
async def get_page(url):
    async with aiohttp.ClientSession() as session:
        #get()、post():
        #headers,params/data,proxy='http://ip:port'
        async with await session.get(url) as response:
            #text()返回字符串形式的响应数据
            #read()返回的二进制形式的响应数据
            #json()返回的就是json对象
            #注意：获取响应数据操作之前一定要使用await进行手动挂起
            page_text = await response.text()
            print(page_text)

tasks = []

for url in urls:
    c = get_page(url)
    task = asyncio.ensure_future(c)
    tasks.append(task)

loop = asyncio.get_event_loop()

loop.run_until_complete(asyncio.wait(tasks))

end = time.time()

print('总耗时:',end-start)

########################################

import requests
from lxml import etree
import time
import os
start = time.time()
headers = {
    'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_12_0) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/72.0.3626.121 Safari/537.36'
}
if not os.path.exists('./libs'):
    os.mkdir('./libs')
url = 'http://pic.netbian.com/4kmeinv/index_%d.html'
a = []
for page in range(2,50):
    new_url = format(url%page)
    page_text = requests.get(url=new_url,headers=headers).text
    tree = etree.HTML(page_text)
    li_list = tree.xpath('//div[@class="slist"]/ul/li')
    for li in li_list:
        img_src = 'http://pic.netbian.com' + li.xpath('./a/img/@src')[0]
        name = img_src.split('/')[-1]
        # data = requests.get(url=img_src).content
        # path = './libs/'+name
        # with open(path,'wb') as fp:
        #     fp.write(data)
        #     print(name,'下载成功')
        a.append(name)
print(len(a))
print('总耗时：',time.time()-start)