python学习Day40--复习+初始协程

一、回顾

1、线程池

　　队列：同一进程内的队列。（先进先出，后进后出，优先级队列）

2、线程池中的回调函数是谁在调用？

　　线程池中的回调函数是子线程调用的，和父线程没有关系。

　　进程池中的回调函数是父进程调用的，和子进程没有关系。

二、协程

1、yield实现状态保存

import time

def func():
    print(123)
    sum = 0
    print(6666)
    yield sum
    print(7777)
    yield sum
    print(8888)
    yield sum

def fff():
    g = func()
    print('这是在ffff函数中')
    print(next(g))
    time.sleep(1)
    print('这是在ffff函数中')
    print(next(g))
    time.sleep(1)
    print('这是在ffff函数中')
    print(next(g))

fff()

2、yield实现并发的假象

　　yield只能实现单纯的切换函数和保存函数状态的功能；不能实现当某一个函数遇到IO阻塞时，自动切换到另一个函数去执行。

　　但是，协程的本质还是主要依靠yield去实现的。

　　如果只拿yield去单纯的实现一个切换假象，你会发现根本没有程序串行执行效率高。

import time

# 在单线程中，如果存在多个函数，如果有某个函数发生IO操作，我想让程序马上切换到另一个函数去执行
#  以此来实现一个假的并发现象。
# 总结：
#    yield 只能实现单纯的切换函数和保存函数状态的功能
#    不能实现：当某一个函数遇到io阻塞时，自动的切换到另一个函数去执行
#    目标是：当某一个函数中遇到IO阻塞时，程序能自动的切换到另一个函数去执行
#            如果能实现这个功能，那么每个函数都是一个协程
#
#    但是  协程的本质还是主要依靠于yield去实现的。
#
#    如果只是拿yield去单纯的实现一个切换的现象，你会发现，跟本没有程序串行执行效率高

def consumer():
    while 1:
        x = yield
        # print(x)

def producer():
    g = consumer()
    next(g)
    for i in range(100000000):
        g.send(i)

start = time.time()
producer()
print('yield:',time.time() - start)



def consumer(l):
    # for i in l:
        # print(i)
    pass

def producer():
    l = []
    for i in range(100000000):
        l.append(i)
    return l

start = time.time()
l = producer()
consumer(l)
print(time.time() - start)

3、协程

　　协程是一个比线程更加轻量级的单位，是组成线程的各个函数。

　　协程本身没有实体。

　　并发的本质：切换+保存状态

 4、greenlet模块

　　能简单的实现函数与函数之间的切换，但是遇到IO操作，不能自动切换到其他函数中。

（1）注册一下函数func，将函数注册成一个对象f1

　　f1 = greenlet(func)

（2）调用func，使用f1.switch( )，如果func需要转参，就在switch这里传即可。

　　grenlet模块可以实现在某函数内部遇到IO操作，就自动的切换到其他函数内部去执行。

from greenlet import greenlet
import time
# greenlet 只是可以实现一个简单的切换功能，还是不能做到遇到IO就切换
# g1 = greenlet(func)   实例化一个对象
# g1.switch()  用这种方式去调用func函数
# 当使用switch调用func的时候，什么时候func会停止运行？
#   1 要么return      2 要么在func内部又遇到 switch

def eat(name):
    print('%s吃炸鸡'%name)
    time.sleep(2)
    f2.switch('小雪2')
    print('%s吃雪糕'%name)
    f2.switch()

def drink(name):
    print('%s喝啤酒'%name)
    f1.switch()
    print('%s喝可乐'%name)


f1 = greenlet(eat)
f2 = greenlet(drink)
f1.switch('小雪')

5、gevent模块

　　可以实现在某函数内部遇到IO操作，就自动的切换到其他函数内部去执行。

　　g = gevent(func)

　　gevent.join(g)   等待g指向的函数func执行完毕，如果在执行过程中，遇到IO，就切换

　　gevent.joinall([g1,g2,g3])  等待g1 g2 g3指向的函数func执行完成。

（1）简单应用

import gevent
import time
# gevent 可以实现  当函数中遇到io操作时，就自动的切换到另一个函数
# g1 = gevent.spawn(func,参数)
# gevent.join(g1) 让func执行完毕
# gevent.joinall([g1,g2,g3,g4])
#    func停止的原因: 1 func执行完了     2 遇到IO操作了

# def func1():
#     print('1 2 3 4')
#     # gevent.sleep(1)
#     time.sleep(1)
#     print('3 2 3 4')
#     # gevent.sleep(1)
#
# def func2():
#     print('2 2 3 4')
#     # gevent.sleep(1)
#     time.sleep(1)# gevent不能识别其他的IO操作，只能识别自己认识的IO
#     print('再来一次')
#
# g1 = gevent.spawn(func1)
# g2 = gevent.spawn(func2)
# g1.join()# 等待g1指向的任务执行结束


#########################################    以下解决gevent不能识别其他IO操作的事情
from gevent import monkey
monkey.patch_all()# 可以让gevent识别大部分常用的IO操作
import time

def func1():
    print('1 2 3 4')
    time.sleep(1)
    print('3 2 3 4')
    # gevent.sleep(1)

def func2():
    print('2 2 3 4')
    time.sleep(1)
    print('再来一次')

g1 = gevent.spawn(func1)
g2 = gevent.spawn(func2)
g1.join()# 等待g1指向的任务执行结束
g2.join()

（2）串行和并发的效率对比

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

def func1(num):
    time.sleep(1)
    print(num)
# 串行
start = time.time()
for i in range(10):
    func1(i)
print(time.time() - start)

# 并发
start = time.time()
l = []
for i in range(10):
    g = gevent.spawn(func1,i)
    l.append(g)
gevent.joinall(l)
print(time.time() - start)

（3）爬虫示例

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

def get_result(url):# 任务函数
    res = requests.get(url)
    print(url,res.status_code,len(res.text))

url_l = ['http://www.baidu.com',
         'https://www.jd.com',
         'http://www.taobao.com',
         'http://www.qq.com',
         'http://www.mi.com',
         'http://www.cnblogs.com']

def sync_func(url_l):
    '''同步调用'''
    for url in url_l:# 串行执行任务函数
        get_result(url)

def async_func(url_l):
    '''异步'''
    l = []
    for url in url_l:
        g = gevent.spawn(get_result,url)# 使用gevent，协程去并发实现执行任务函数
        # 当遇见请求某个网页发生比较大的网络延迟（IO），马上会切换到其他的任务函数
        l.append(g)
    gevent.joinall(l)# 等待所有任务函数执行结束

start = time.time()
sync_func(url_l)
print('sync:',time.time() - start)

start = time.time()
async_func(url_l)
print('async:',time.time() - start)

【问题】为什么要有协程？

　　因为想要在单线程内实现并发的效果。

　　因为Cpython有GIL锁，限制了在同一个时间点只能执行一个线程。

　　所以想要在执行一个线程的期间，充分的利用CPU的性能。

　　所以才有了想在线程内实现并发的效果。

6、IO多路复用（解决问题的思想）

 （1）用非阻塞IO解决阻塞IO问题

import socket

sk = socket.socket()
sk.setblocking(False)
sk.bind(('127.0.0.1',8080))
sk.listen()

l = []
del_l = []
while 1:
    try:
        conn,addr = sk.accept()# 如果是阻塞IO模型，在这里程序会一直等待。
        l.append(conn)# 将每个请求连接的客户端的conn添加到列表中
    except BlockingIOError:
        for conn in l:# 去遍历所有客户端的conn，看看有没有客户端给我发送数据了

            try:
                info = conn.recv(1024).decode('utf-8')# 尝试接收，看看有没有客户端给我发数据
                if not info:# 如果客户端正常执行了close，服务器会接收到一个空
                    del_l.append(conn)# 将已经结束的客户端的conn，添加到要删除的列表中
                    print('客户端正常退出了!')
                    conn.close()# 因为客户端已经主动close，所以服务器端的conn也要close
                else:
                    print(info)
                    conn.send(info.upper().encode('utf-8'))
            except BlockingIOError:
                continue# 是没有接受到客户端发来的数据而报错
            except ConnectionResetError:
                pass# 是因为客户端强制退出而报错
        if del_l:
            for conn in del_l:
                l.remove(conn)
            del_l = []# 在删除完主动关闭的客户端的连接之后，应该把此列表清空，否则报错

import socket
sk = socket.socket()
sk.connect(('127.0.0.1',8080))

while 1:
    msg_s = input('>>>')
    if not msg_s:continue
    if msg_s == 'q':break
    sk.send(msg_s.encode('utf-8'))
    print(sk.recv(1024).decode('utf-8'))
sk.close()

【不推荐使用】

　　循环使用recv( )将大幅占用CPU占用率；任务完成的响应延迟增大了。

（2）多路IO复用解决阻塞IO问题

import select
import socket

sk = socket.socket()
sk.bind(('127.0.0.1',8888))
sk.listen()
del_l = []
rlist = [sk]# 是用来让select帮忙监听的 所有 接口
# select：windows/linux是监听事件有没有数据到来
# poll:  linux   也可以做select的工作
# epoll: linux   也可以做类似的工作
while 1:
    r,w,x = select.select(rlist,[],[])# 传参给select，当rlist列表中哪个接口有反应，就返回给r这个列表
    if r:
        for i in r:# 循环遍历r，看看有反应的接口到底是sk  还是conn
            if i == sk:
                # 如果是sk，那就表示有客户端的连接请求
                '''sk有数据要接收，代表着有客户端要来连接'''
                conn,addr = i.accept()
                rlist.append(conn)# 把新的客户端的连接，添加到rlist，继续让select帮忙监听
            else:
                # 如果是conn，就表示有客户端给我发数据了
                '''conn有数据要接收，代表要使用recv'''
                try:
                    msg_r = i.recv(1024).decode('utf-8')
                    if not msg_r:
                        '''客户端执行了close，客户端主动正常关闭连接'''
                        del_l.append(i)
                        i.close()
                    else:
                        print(msg_r)
                        i.send(msg_r.upper().encode('utf-8'))
                except ConnectionResetError:
                    pass
        if del_l:# 删除那些主动断开连接的客户端的conn
            for conn in del_l:
                rlist.remove(conn)
            del_l.clear()

import socket
sk = socket.socket()
sk.connect(('127.0.0.1',8888))

while 1:
    msg_s = input('>>>')
    if not msg_s:continue
    if msg_s == 'q':break
    sk.send(msg_s.encode('utf-8'))
    print(sk.recv(1024).decode('utf-8'))
sk.close()

（3）异步IOpython实现不了

（4）select与poll和epoll的区别

　　select和poll有一个共同的机制，都是采用轮询的方式询问内核，有没有数据准备好了。

　　select有一个最大监听事件的限制，32位机限制1024，64位机限制2048；poll没有，理论上poll可以开启无限大，1G内存大概够开启10W个事件去监听。

　　epoll是最好的，采用的是回调机制，解决了select和poll共同存在的问题，而且epoll理论上也可以开启无限多个监听事件。