Python知识点-协程

Posted 2021-01-28 ghx1

协程就是一个线程，只是说再一个线程上来回切换。

协程切换任务是靠代码，遇到IO 操作就切换，而线程和进程是靠操作系统自动切换

 1.greenlet

from greenlet import greenlet

def Producer():
    while True:
        print(‘我是生产者我会生产大肉包‘)
        time.sleep(1)
        c.switch()  # 切换到消费者
        print(‘生产结束‘)
def Consumer():
    while True:
        print(‘我是消费者我就会吃‘)
        time.sleep(1)
        p.switch()  # 切换到生产者
        print(‘消费结束‘)


# 将普通函数编程协程
c = greenlet(Consumer)
p = greenlet(Producer)
c.switch() #consumer先执行

2.gevent 只有协程遇到能识别的IO操作才切换(from gevent import monkey;monkey.patch_all())

# 将python标准库中的一些阻塞操作变为非阻塞
from gevent import monkey;monkey.patch_all()
# 使用猴子补丁要写在第一行
import gevent

def test1():
    print("test1")
    gevent.sleep(1)  # 模拟耗时操作
    print("test11")

def test2():
    print("test2")
    gevent.sleep(1)  # 模拟耗时操作
    print("test22")

g1 = gevent.spawn(test1)  # 将函数封装成协程，并启动
g2 = gevent.spawn(test2)
gevent.joinall([g1, g2])

greenlet 和gevent 区别在于一个是手动切换，一个是自动切换，gevent是在greenlet的基础上实现的。

# 基于gevent的并发服务器实现
import gevent
# 将python内置的socket换成封装了IO多路复用的socket
from  gevent import monkey;monkey.patch_all()
import socket

# 实例化socket
server = socket.socket()
# 绑定ip和端口
server.bind((‘0.0.0.0‘, 8000))
# 绑定监听数量
server.listen(1000)

def worker(conn):

    while True:
        recv_data = conn.recv(1024)  # 等待接收数据
        if recv_data:
            print(recv_data)
            conn.send(recv_data))  # 将接收的数据原路返回
        else:
            conn.close()  # 发送完毕断开
            break

while True:
    conn, addr = server.accept()  # 等待客户端连接，遇到阻塞切换
    gevent.spawn(worker, conn)  # 生成协程，并将conn作为参数传入

 

3.asyncio py3.4开始加入的内置标准库 使用yield from 切换协程，遇到阻塞就切换，等阻塞结束后拿到返回值

import threading
import asyncio

@asyncio.coroutine
def hello():
    print(‘Hello world! (%s)‘ % threading.currentThread())
    yield from asyncio.sleep(1) #模拟创建一个1秒后完成的协程
    print(‘Hello again! (%s)‘ % threading.currentThread())

#获取event_loop
loop = asyncio.get_event_loop()
tasks = [hello(), hello()]
#执行协程
loop.run_until_complete(asyncio.wait(tasks))
loop.close() #关闭事件循环

结果：

Hello world! (<_MainThread(MainThread, started 140735195337472)>)
Hello world! (<_MainThread(MainThread, started 140735195337472)>)
(暂停约1秒)
Hello again! (<_MainThread(MainThread, started 140735195337472)>)
Hello again! (<_MainThread(MainThread, started 140735195337472)>)

由打印的当前线程名称可以看出，两个coroutine是由同一个线程并发执行的。

如果把asyncio.sleep()换成真正的IO操作，则多个coroutine就可以由一个线程并发执行。

我们用asyncio的异步网络连接来获取sina、sohu和163的网站首页：

import asyncio

@asyncio.coroutine
def wget(host):
    print(‘wget %s...‘ % host)
    connect = asyncio.open_connection(host, 80) #连接服务器，创建一个socket，返回两个对应的流控制对象StreamReader、StreamWriter。
    reader, writer = yield from connect 
    header = ‘GET / HTTP/1.0
Host: %s

‘ % host
    #调用写对象write函数来发送数据给服务器，但是这里并没有直正把数据发送出去，只是写到内部缓冲区，
    writer.write(header.encode(‘utf-8‘))
    #调用writer.drain()函数就是等着socket把数据发送出去
    yield from writer.drain()
    while True:
        #接收数据的无限循环，从服务器接受数据，无数据接收到，就结束循环。
        line = yield from reader.readline()
        if line == b‘
‘:
            break
        print(‘%s header > %s‘ % (host, line.decode(‘utf-8‘).rstrip()))
    # Ignore the body, close the socket
    writer.close()

loop = asyncio.get_event_loop() #创建 event_loop
tasks = [wget(host) for host in [‘www.sina.com.cn‘, ‘www.sohu.com‘, ‘www.163.com‘]]
loop.run_until_complete(asyncio.wait(tasks))
loop.close()

4.async/await  是新写法，await替换yield from 和 async 替换@asyncio.coroutine，其他不变

import threading
import asyncio


async def hello():
    print(‘Hello world! (%s)‘ % threading.currentThread())
    await asyncio.sleep(1) #模拟创建一个1秒后完成的协程
    print(‘Hello again! (%s)‘ % threading.currentThread())

#获取event_loop
loop = asyncio.get_event_loop()
tasks = [hello(), hello()]
#执行协程
loop.run_until_complete(asyncio.wait(tasks))
loop.close() #关闭事件循环