您的位置: 首页  >  文章  >  线程下的协程

线程下的协程

分类: 文章 • 2024-06-08 18:13:34

一      什么是协程

协程，又可称之为微线程。

线程下的协程

协程的特点在于是一个线程在执行，相对于线程而言，具有一定的优势：

1）   协程极高的执行效率。因为子程序切换不是线程切换，而是由程序自身控制，因此，没有线程切换的开销，和多线程比，线程数量越多，协程的性能优势就越明显

2）    不需要多线程的锁机制，因为只有一个线程，也不存在同时写变量冲突，在协程中控制共享资源不加锁，只需要判断状态就好了，所以执行效率比多线程高很多

因为协程是一个线程执行，就如何高效的利用CPU而言，最简单的方法是多进程+协程，既充分利用多核，又充分发挥协程的高效率，可获得极高的性能

二     协程的实现

1）    通过yield实现协程

传统的生产者-消费者模型是一个线程写消息，一个线程取消息，通过锁机制控制队列和等待，但一不小心就可能死锁。

如果改用协程，生产者生产消息后，直接通过yield跳转到消费者开始执行，待消费者执行完毕后，切换回生产者继续生产，效率极高

import threading
import time
def producer(c):
    c.__next__()
    n=0
    while n<5:
        n+=1
        print('[producer] %s ...'%(n))
        res=c.send(n)
        print('[consumer] return %s'%(res))
def consumer():
    r='a'
    while True:
        n=yield r
        if not n:
            return
        print('[consumer] %s ...'%(n))
        time.sleep(1)
        r='200 OK'
if __name__ == '__main__':
    print('活跃线程数：',threading.active_count())
    c=consumer()
    producer(c)
    print('活跃线程数：',threading.active_count())

执行结果：

线程下的协程

注意到consumer函数是一个generator（生成器），把一个consumer传入produce后：

  1. 首先调用c.next()启动生成器

  2. 然后，一旦生产了东西，通过c.send(n)切换到consumer执行；

  3. consumer通过yield拿到消息，处理，又通过yield把结果传回；

  4. produce拿到consumer处理的结果，继续生产下一条消息；

  5. produce决定不生产了，通过c.close()关闭consumer，整个过程结束。

整个流程无锁，由一个线程执行，produce和consumer协作完成任务，所以称为“协程”，而非线程的抢占式多任务

2）     通过genvent实现协程

import time
from gevent import monkey
monkey.patch_all()
import gevent
def job(n):
    for i in range(n):
        print(gevent.getcurrent(),i)
        time.sleep(1)
def main():
    g1=gevent.spawn(job,2)
    g2=gevent.spawn(job,3)
    g3=gevent.spawn(job,1)
 
    gevent.joinall([g1,g2,g3])

    print('任务结束...')

main()

执行结果：

线程下的协程

注意到协程之间是交互执行的，而不是依次执行

另外  from gevent import monkey 是为gevent函数打补丁

三      协程案例

比较多线程与多协程处理网页的效率

from concurrent.futures import ThreadPoolExecutor
from urllib.request import urlopen

import gevent
from gevent import monkey

from mytime import timeit

monkey.patch_all()

def load_url(url):
    with urlopen(url) as f:
        f.read()
URLS=['http://httpbin.org','http://example.com/']*20
@timeit
def gevent_main():
    gevents=[gevent.spawn(load_url,url) for url in URLS]
    gevent.joinall(gevents)
@timeit
def thread_main():
    with ThreadPoolExecutor(max_workers=100) as f:
        f.map(load_url,URLS)
if __name__ == '__main__':
    gevent_main()
    thread_main()

线程数量越多，协程的性能优势就越明显

四     客户端与服务端

C/S方式， 客户端发送一条命令， 服务端返回命令的执行结果；ssh-client, ssh-server
        - ssh
        - paramiko----(socket进行封装)

1).    服务端

import os
import socket
server = socket.socket()
server.bind(('172.25.254.69',9001))
server.listen()
print("服务端已经启动9001端口....")
sockobj,address = server.accept()
while True:
    recv_data = sockobj.recv(1024).decode('utf-8')
    if recv_data == 'quit':
        break
    res = os.popen(recv_data).read()
    sockobj.send(res.encode('utf-8'))
sockobj.close()
server.close()

执行的结果：

线程下的协程

客户端：

import socket
HOST='172.25.254.69'
PORT = 9001
client = socket.socket()
client.connect((HOST,PORT))
while True:
    send_data = input('[[email protected]]:')
    client.send(send_data.encode('utf-8'))
    if send_data == 'quit':
        break
    recv_data = client.recv(1024).decode('utf-8')
    print('[[email protected]]:%s'%(recv_data))
client.close()

执行的结果：

线程下的协程

 

相关推荐