协程
协程,又称微线程,纤程。英文名Coroutine。一句话说明什么是线程:协程是一种用户态的轻量级线程。
协程拥有自己的寄存器上下文和栈。协程调度切换时,将寄存器上下文和栈保存到其他地方,在切回来的时候,恢复先前保存的寄存器上下文和栈。因此:
协程能保留上一次调用时的状态(即所有局部状态的一个特定组合),每次过程重入时,就相当于进入上一次调用的状态,换种说法:进入上一次离开时所处逻辑流的位置。
协程的好处:
- 无需线程上下文切换的开销
- 无需原子操作锁定及同步的开销
- "原子操作(atomic operation)是不需要synchronized",所谓原子操作是指不会被线程调度机制打断的操作;这种操作一旦开始,就一直运行到结束,中间不会有任何 context switch (切换到另一个线程)。原子操作可以是一个步骤,也可以是多个操作步骤,但是其顺序是不可以被打乱,或者切割掉只执行部分。视作整体是原子性的核心。
- 方便切换控制流,简化编程模型
- 高并发+高扩展性+低成本:一个CPU支持上万的协程都不是问题。所以很适合用于高并发处理。
缺点:
- 无法利用多核资源:协程的本质是个单线程,它不能同时将 单个CPU 的多个核用上,协程需要和进程配合才能运行在多CPU上.当然我们日常所编写的绝大部分应用都没有这个必要,除非是cpu密集型应用。
- 进行阻塞(Blocking)操作(如IO时)会阻塞掉整个程序
使用yield实现协程操作例子
1 import time 2 import queue 3 def consumer(name): 4 print("--->starting eating baozi...") 5 while True: 6 new_baozi = yield 7 print("[%s] is eating baozi %s" % (name,new_baozi)) 8 #time.sleep(1) 9 10 def producer(): 11 12 r = con.__next__() 13 r = con2.__next__() 14 n = 0 15 while n < 5: 16 n +=1 17 con.send(n) 18 con2.send(n) 19 print("\033[32;1m[producer]\033[0m is making baozi %s" %n ) 20 21 22 if __name__ == '__main__': 23 con = consumer("c1") 24 con2 = consumer("c2") 25 p = producer()
看楼上的例子,我问你这算不算做是协程呢?你说,我他妈哪知道呀,你前面说了一堆废话,但是并没告诉我协程的标准形态呀,我腚眼一想,觉得你说也对,那好,我们先给协程一个标准定义,即符合什么条件就能称之为协程:
- 必须在只有一个单线程里实现并发
- 修改共享数据不需加锁
- 用户程序里自己保存多个控制流的上下文栈
- 一个协程遇到IO操作自动切换到其它协程
基于上面这4点定义,我们刚才用yield实现的程并不能算是合格的线程,因为它有一点功能没实现,哪一点呢?
Greenlet
greenlet是一个用C实现的协程模块,相比与python自带的yield,它可以使你在任意函数之间随意切换,而不需把这个函数先声明为generator
1 # -*- coding:utf-8 -*- 2 3 4 from greenlet import greenlet 5 6 7 def test1(): 8 print(12) 9 gr2.switch() 10 print(34) 11 gr2.switch() 12 13 14 def test2(): 15 print(56) 16 gr1.switch() 17 print(78) 18 19 20 gr1 = greenlet(test1) 21 gr2 = greenlet(test2) 22 gr1.switch()
感觉确实用着比generator还简单了呢,但好像还没有解决一个问题,就是遇到IO操作,自动切换,对不对?
Gevent
Gevent 是一个第三方库,可以轻松通过gevent实现并发同步或异步编程,在gevent中用到的主要模式是Greenlet, 它是以C扩展模块形式接入Python的轻量级协程。 Greenlet全部运行在主程序操作系统进程的内部,但它们被协作式地调度。
1 import gevent 2 3 def func1(): 4 print('\033[31;1m李闯在跟海涛搞...\033[0m') 5 gevent.sleep(2) 6 print('\033[31;1m李闯又回去跟继续跟海涛搞...\033[0m') 7 8 def func2(): 9 print('\033[32;1m李闯切换到了跟海龙搞...\033[0m') 10 gevent.sleep(1) 11 print('\033[32;1m李闯搞完了海涛,回来继续跟海龙搞...\033[0m') 12 13 14 gevent.joinall([ 15 gevent.spawn(func1), 16 gevent.spawn(func2), 17 #gevent.spawn(func3), 18 ])
输出:
李闯在跟海涛搞...
李闯切换到了跟海龙搞...
李闯搞完了海涛,回来继续跟海龙搞...
李闯又回去跟继续跟海涛搞...
同步与异步的性能区别
1 import gevent 2 3 def task(pid): 4 """ 5 Some non-deterministic task 6 """ 7 gevent.sleep(0.5) 8 print('Task %s done' % pid) 9 10 def synchronous(): 11 for i in range(1,10): 12 task(i) 13 14 def asynchronous(): 15 threads = [gevent.spawn(task, i) for i in range(10)] 16 gevent.joinall(threads) 17 18 print('Synchronous:') 19 synchronous() 20 21 print('Asynchronous:') 22 asynchronous()
上面程序的重要部分是将task函数封装到Greenlet内部线程的gevent.spawn
。 初始化的greenlet列表存放在数组threads
中,此数组被传给gevent.joinall
函数,后者阻塞当前流程,并执行所有给定的greenlet。执行流程只会在 所有greenlet执行完后才会继续向下走。
遇到IO阻塞时会自动切换任务
1 from gevent import monkey; monkey.patch_all() 2 import gevent 3 from urllib.request import urlopen 4 5 def f(url): 6 print('GET: %s' % url) 7 resp = urlopen(url) 8 data = resp.read() 9 print('%d bytes received from %s.' % (len(data), url)) 10 11 gevent.joinall([ 12 gevent.spawn(f, 'https://www.python.org/'), 13 gevent.spawn(f, 'https://www.yahoo.com/'), 14 gevent.spawn(f, 'https://github.com/'), 15 ])
通过gevent实现单线程下的多socket并发
server side
1 import sys 2 import socket 3 import time 4 import gevent 5 6 from gevent import socket,monkey 7 monkey.patch_all() 8 9 10 def server(port): 11 s = socket.socket() 12 s.bind(('0.0.0.0', port)) 13 s.listen(500) 14 while True: 15 cli, addr = s.accept() 16 gevent.spawn(handle_request, cli) 17 18 19 20 def handle_request(conn): 21 try: 22 while True: 23 data = conn.recv(1024) 24 print("recv:", data) 25 conn.send(data) 26 if not data: 27 conn.shutdown(socket.SHUT_WR) 28 29 except Exception as ex: 30 print(ex) 31 finally: 32 conn.close() 33 if __name__ == '__main__': 34 server(8001)
client side
1 import socket 2 3 HOST = 'localhost' # The remote host 4 PORT = 8001 # The same port as used by the server 5 s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 6 s.connect((HOST, PORT)) 7 while True: 8 msg = bytes(input(">>:"),encoding="utf8") 9 s.sendall(msg) 10 data = s.recv(1024) 11 #print(data) 12 13 print('Received', repr(data)) 14 s.close()
1 import socket 2 import threading 3 4 def sock_conn(): 5 6 client = socket.socket() 7 8 client.connect(("localhost",8001)) 9 count = 0 10 while True: 11 #msg = input(">>:").strip() 12 #if len(msg) == 0:continue 13 client.send( ("hello %s" %count).encode("utf-8")) 14 15 data = client.recv(1024) 16 17 print("[%s]recv from server:" % threading.get_ident(),data.decode()) #结果 18 count +=1 19 client.close() 20 21 22 for i in range(100): 23 t = threading.Thread(target=sock_conn) 24 t.start()