网络编程--协程

目录

• 引子
• 1、单线程实现并发
  • 是否可行
  • 如何够实现
  • greenlet模块实现并发
• 2、协程
  • gevent模块

引子

上一节中咱们知道GIL锁将致使CPython中多线程没法并行执行，只能并发的执行。

而并发实现的原理是切换+保存,那就意味着使用多线程实现并发,就须要为每个任务建立一个线程,必然增长了线程建立销毁与切换的带来的开销

明显的问题就是,高并发状况下,因为任务数量太多致使没法开启新的线程,使得即没有实际任务要执行,也没法建立新线程来处理新任务的状况

如何解决上述问题呢,首先要保证并发效果,而后来想办法避免建立线程带来的开销问题;

协程既是所以而出现的,其原理是使用单线程来实现多任务并发,那么如何能实现单线程并发呢?

1、单线程实现并发

是否可行

单线程实现并发这句话乍一听好像在瞎说

首先须要明确并发的定义

并发：指的是多个任务同时发生，看起来好像是同时都在进行

并行：指的是多个任务真正的同时进行

早期的计算机只有一个CPU，既然CPU能够切换线程来实现并发，那么为什么不能在线程中切换任务来并发呢？

因此线程实现并发理论上是可行的

如何够实现

并发 = 切换任务+保存状态，只要找到一种方案，可以在两个任务之间切换执行而且保存状态，那就能够实现单线程并发

python中的生成器就具有这样一个特色，每次调用next都会回到生成器函数中执行代码，这意味着任务之间能够切换，而且是基于上一次运行的结果，这意味着生成器会自动保存执行状态！

因而乎咱们能够利用生成器来实现并发执行：

def task1():
    while True:
        yield
        print("task1 run")

def task2():
    g = task1()
    while True:
        next(g)
        print("task2 run")
task2()

并发虽然实现了，可是这对效率的影响是好是坏呢？来测试一下

# 两个计算任务一个采用生成器切换并发执行  一个直接串行调用
import  time
def task1():
    a = 0
    for i in range(10000000):
        a += i
        yield

def task2():
    g = task1()
    b = 0
    for i in range(10000000):
        b += 1
        next(g)
s = time.time()
task2()
print("并发执行时间",time.time()-s)

# 单线程下串行执行两个计算任务 效率反而比并发高 由于并发须要切换和保存
def task1():
    a = 0
    for i in range(10000000):
        a += i
def task2():
    b = 0
    for i in range(10000000):
        b += 1
s = time.time()
task1()
task2()
print("串行执行时间",time.time()-s)

能够看到对于纯计算任务而言，单线程并发反而使执行效率降低了一半左右，因此这样的方案对于纯计算任务而言是没有必要的

greenlet模块实现并发

咱们暂且不考虑这样的并发对程序的好处是什么，在上述代码中，使用yield来切换是的代码结构很是混乱，若是十个任务须要切换呢，不敢想象！所以就有人专门对yield进行了封装，这便有了greenlet模块

from greenlet import greenlet

def eat(name):
    print('%s eat 1' %name)
    g2.switch('jack')
    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('rose')#能够在第一次switch时传入参数，之后都不须要再次传

该模块简化了yield复杂的代码结构，实现了单线程下多任务并发，可是不管直接使用yield仍是greenlet都不能检测IO操做，遇到IO时一样进入阻塞状态，一样的对于纯计算任务而言效率也是没有任何提高的。

测试:

#切换
from greenlet import greenlet
import time
def f1():
    res=1
    for i in range(100000000):
        res+=i
        g2.switch()

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

start=time.time()
g1=greenlet(f1)
g2=greenlet(f2)
g1.switch()
stop=time.time()
print('run time is %s' %(stop-start)) # 52.763017892837524

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

任务的代码一般会既有计算操做又有阻塞操做，咱们彻底能够在执行任务1时遇到阻塞，就利用阻塞的时间去执行任务2。。。。如此，才能提升效率，这就用到了Gevent模块。

2、协程

协程：是单线程下的并发，又称微线程，纤程。英文名Coroutine。是一种用户态的轻量级线程，即协程是由用户程序本身控制调度的。

须要强调的是：

#1. python的线程属于内核级别的，即由操做系统控制调度（如单线程遇到io或执行时间过长就会被迫交出cpu执行权限，切换其余线程运行）
#2. 单线程内开启协程，一旦遇到io，就会从应用程序级别（而非操做系统）控制切换，以此来提高效率（！！！非io操做的切换与效率无关）

对比操做系统控制线程的切换，用户在单线程内控制协程的切换

优势以下：

#1. 协程的切换开销更小，属于程序级别的切换，操做系统彻底感知不到，于是更加轻量级
#2. 单线程内就能够实现并发的效果，最大限度地利用cpu

缺点以下：

#1. 协程的本质是单线程下，没法利用多核，能够是一个程序开启多个进程，每一个进程内开启多个线程，每一个线程内开启协程来尽量提升效率
#2. 协程本质是单个线程，于是一旦协程出现阻塞，将会阻塞整个线程

gevent模块

Gevent 是一个第三方库，能够轻松经过gevent实现并发编程，在gevent中用到的主要模式是Greenlet, 它是以C扩展模块形式接入Python的轻量级协程。 Greenlet所有运行在主程序操做系统进程的内部，但它们被协做式地调度。

经常使用方法:

#用法
#建立一个协程对象g1，
g1=gevent.spawn(func,1,,2,3,x=4,y=5)
#spawn括号内第一个参数是函数名，如eat，后面能够有多个参数，能够是位置实参或关键字实参，都是传给函数eat的
g2=gevent.spawn(func2)

g1.join() #等待g1结束

g2.join() #等待g2结束

#或者上述两步合做一步：gevent.joinall([g1,g2])

g1.value#拿到func1的返回值

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

import gevent,sys
from gevent import monkey # 导入monkey补丁
monkey.patch_all() # 打补丁 
import time

print(sys.path)

def task1():
    print("task1 run")
    # gevent.sleep(3)
    time.sleep(3)
    print("task1 over")

def task2():
    print("task2 run")
    # gevent.sleep(1)
    time.sleep(1)
    print("task2 over")

g1 = gevent.spawn(task1)
g2 = gevent.spawn(task2)
#gevent.joinall([g1,g2])
g1.join()
g2.join()
# 执行以上代码会发现不会输出任何消息
# 这是由于协程任务都是以异步方式提交，因此主线程会继续往下执行，而一旦执行完最后一行主线程也就结束了，
# 致使了协程任务没有来的及执行，因此这时候必须join来让主线程等待协程任务执行完毕   也就是让主线程保持存活
# 后续在使用协程时也须要保证主线程一直存活，若是主线程不会结束也就意味着不须要调用join

须要注意：

1.若是主线程结束了 协程任务也会当即结束。

2.monkey补丁的原理是把原始的阻塞方法替换为修改后的非阻塞方法，即偷梁换柱，来实现IO自动切换

必须在打补丁后再使用相应的功能，避免忘记，建议写在最上方

咱们能够用threading.current_thread().getName()来查看每一个g1和g2，查看的结果为DummyThread-n，即假线程

monke补丁原理

#myjson.py
def dump():
    print("一个被替换的 dump函数")

def load():
    print("一个被替换的 load函数")

# test.py
import myjson
import json
# 补丁函数
def monkey_pacth_json():
    json.dump = myjson.dump
    json.load = myjson.load
    
# 打补丁
monkey_pacth_json()

# 测试 
json.dump()
json.load()
# 输出：
# 一个被替换的 dump函数
# 一个被替换的 load函数

使用Gevent案例一 爬虫:

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

def get_page(url):
    print('GET: %s' %url)
    response=requests.get(url)
    if response.status_code == 200:
        print('%d bytes received from %s' %(len(response.text),url))


start_time=time.time()
gevent.joinall([
    gevent.spawn(get_page,'https://www.python.org/'),
    gevent.spawn(get_page,'https://www.yahoo.com/'),
    gevent.spawn(get_page,'https://github.com/'),
])
stop_time=time.time()
print('run time is %s' %(stop_time-start_time))

使用Gevent案例二 TCP:

服务器

#=====================================服务端
from gevent import monkey;monkey.patch_all()
from socket import *
import gevent

#若是不想用money.patch_all()打补丁,能够用gevent自带的socket
# from gevent import socket
# s=socket.socket()

def server(server_ip,port):
    s=socket(AF_INET,SOCK_STREAM)
    s.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
    s.bind((server_ip,port))
    s.listen(5)
    while True:
        conn,addr=s.accept()
        gevent.spawn(talk,conn,addr)

def talk(conn,addr):
    try:
        while True:
            res=conn.recv(1024)
            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 threading import Thread
from socket import *
import threading

def client(server_ip,port):
    c=socket(AF_INET,SOCK_STREAM) #套接字对象必定要加到函数内，即局部名称空间内，放在函数外则被全部线程共享，则你们公用一个套接字对象，那么客户端端口永远同样了
    c.connect((server_ip,port))

    count=0
    while True:
        c.send(('%s say hello %s' %(threading.current_thread().getName(),count)).encode('utf-8'))
        msg=c.recv(1024)
        print(msg.decode('utf-8'))
        count+=1
if __name__ == '__main__':
    for i in range(500):
        t=Thread(target=client,args=('127.0.0.1',8080))
        t.start()