第一章 Python 线程

1、线程前戏

(1) 并发和并行：

• 并发：伪，由于执行速度特别快，感觉不到停顿

• 并行：真，创建10个人同时操作

(2) 线程，进程

2-1 单进程，单线程的应用程序

print("hello world!!!")

2-2 到底什么是线程？什么是进程？

python自己是没有线程和进程的，python中调用的操作系统的线程和进程

2-3 单进程，多线程的应用程序

import threading
?
print("hello world!!!")
?
def func(arg):
    print(arg)
    
# 创建线程
t = threading.Thread(target=func,args=(11,))
t.start()
?
print(123)

一个应用程序（软件），可以有多个进程（默认只有一个），一个进程中也可以创建多个线程（默认一个）

 

2-4 Java or C# 中的进程与线程

 

2-5 Python 中的进程和线程

 

2-6 总结

操作系统帮助开发者操作硬件。

程序员写好代码在操作系统上运行（依赖解释器）。

当任务特别多时以前你的代码：

import requests
import uuid
?
url_list = [
    'http://pic.sc.chinaz.com/Files/pic/pic9/202008/apic27292_s.jpg',
    'http://pic.sc.chinaz.com/Files/pic/pic9/202008/apic27262_s.jpg',
    'http://pic1.sc.chinaz.com/Files/pic/pic9/202008/apic27250_s.jpg',
]
?
def task(url):
    """
    1. DNS解析，根据域名解析出IP
    2. 创建socket客户端    sk = socket.socket()
    3. 向服务端发起连接请求 sk.connect()
    4. 发送数据（我要图片） sk.send(...)
    5. 接收数据            sk.recv(8096)
    接收到数据后写入文件。
    """
    ret = requests.get(url)
    file_name = str(uuid.uuid4()) + '.jpg'
    with open(file_name, mode='wb') as f:
        f.write(ret.content)
?
for url in url_list:
    task(url)
"""
- 你写好代码
- 交给解释器运行： python s1.py 
- 解释器读取代码，再交给操作系统去执行，根据你的代码去选择创建多少个线程/进程去执行（单进程/单线程）。
- 操作系统调用硬件：硬盘、cpu、网卡....
"""

当任务特别多时现在你的代码：

import threading
import requests
import uuid
?
url_list = [
    'http://pic.sc.chinaz.com/Files/pic/pic9/202008/apic27292_s.jpg',
    'http://pic.sc.chinaz.com/Files/pic/pic9/202008/apic27262_s.jpg',
    'http://pic1.sc.chinaz.com/Files/pic/pic9/202008/apic27250_s.jpg',
]
?
def task(url):
    """
    1. DNS解析，根据域名解析出IP
    2. 创建socket客户端    sk = socket.socket()
    3. 向服务端发起连接请求 sk.connect()
    4. 发送数据（我要图片） sk.send(...)
    5. 接收数据            sk.recv(8096)
    接收到数据后写入文件。
    """
    ret = requests.get(url)
    file_name = str(uuid.uuid4()) + '.jpg'
    with open(file_name, mode='wb') as f:
        f.write(ret.content)
?
for url in url_list:
    t = threading.Thread(target=task, args=(url,))
    t.start()
?
"""
- 你写好代码
- 交给解释器运行： python s2.py 
- 解释器读取代码，再交给操作系统去执行，根据你的代码去选择创建多少个线程/进程去执行（单进程/4线程）。
- 操作系统调用硬件：硬盘、cpu、网卡....
"""

Python多线程情况下：

• 计算密集型操作：效率低。（GIL锁）

• IO操作： 效率高

Python多进程的情况下：

• 计算密集型操作：效率高（浪费资源）。 不得已而为之。

• IO操作： 效率高 （浪费资源）。

以后写Python时：

• IO密集型用多线程： 文件/输入输出/socket网络通信

• 计算密集型用多进程。

扩展：

• Java多线程情况下：

  • 计算密集型操作：效率高。

  • IO操作： 效率高

• Python多进程的情况下：

  • 计算密集型操作：效率高（浪费资源）。

  • IO操作： 效率高 浪费资源）。

 

(3) Python中线程和进程（GIL锁）

• Python内置的一个全局解释器锁，锁的作用就是保证同一时刻一个进程中只有一个线程可以被cpu调度。

• 为什么有这把GIL锁？

  Python语言的创始人在开发这门语言时，目的快速把语言开发出来，如果加上GIL锁（C语言加锁），切换时按照100条字节指令来进行线程间的切换。

• GIL锁，全局解释器锁。用于限制一个进程中同一时刻只有一个线程被cpu调度。

• 扩展：默认GIL锁在执行100个cpu指令（过期时间）。

(4) Python线程和进程之间的区别

• 线程，cpu工作的最小单元。

• 进程，为线程提供一个资源共享的空间。

• 一个进程中默认是有一个主线程。

 

2、Python 初识线程

(1) 线程的基本使用

import threading
?
def func(arg):
    print(arg)
    
# 创建线程
t = threading.Thread(target=func,args=(11,))
t.start()
?
print(123)

运行结果:

11
123

(2) 主线程默认等子线程执行完毕

import threading
import time
?
def func(arg):
    time.sleep(arg)
    print(arg)
?
# 创建线程t1
t1 = threading.Thread(target=func, args=(3,))
t1.start()
?
# 创建线程t2
t2 = threading.Thread(target=func, args=(9,))
t2.start()
?
print(123)

运行结果:

123
3
9

(3) 主线程不再等，主线程终止则所有子线程终止

import threading
import time
?
def func(arg):
    time.sleep(2)
    print(arg)
?
t1 = threading.Thread(target=func, args=(3,))
t1.setDaemon(True)
t1.start()
?
t2 = threading.Thread(target=func, args=(9,))
t2.setDaemon(True)
t2.start()
?
print(123)

运行结果：

123

(4) 开发者可以控制主线程等待子线程（最多等待时间）

import threading
import time
?
def func(arg):
    time.sleep(0.01)
    print(arg)
?
print('创建子线程t1')
t1 = threading.Thread(target=func, args=(3,))
t1.start()
# 无参数，让主线程在这里等着，等到子线程t1执行完毕，才可以继续往下走。
# 有参数，让主线程在这里最多等待n秒，无论是否执行完毕，会继续往下走。
t1.join(2)
?
print('创建子线程t2')
t2 = threading.Thread(target=func, args=(9,))
t2.start()
t2.join(2)  # 让主线程在这里等着，等到子线程t2执行完毕，才可以继续往下走。
?
print(123)

运行结果：

创建子线程t1
3
创建子线程t2
9
123

(5) 线程名称

import threading
?
def func(arg):
    # 获取当前执行该函数的线程的对象
    t = threading.current_thread()
    # 根据当前线程对象获取当前线程名称
    name = t.getName()
    print(name, arg)
?
t1 = threading.Thread(target=func, args=(11,))
t1.setName('线程1')
t1.start()
?
t2 = threading.Thread(target=func, args=(22,))
t2.setName('线程2')
t2.start()
?
print(123)

运行结果：

线程1 11
线程2 22
123

(6) 线程本质

import threading
?
# 先打印：11？123？
def func(arg):
    print(arg)
?
t1 = threading.Thread(target=func, args=(11,))
t1.start()
# start 是开始运行线程吗？不是
# start 告诉cpu，我已经准备就绪，你可以调度我了。
print(123)

运行结果：

11
123

(7) 补充：面向对象版本的多线程

import threading
?
# 多线程方式：1 (常见)
def func(arg):
    print(arg)
    
t1 = threading.Thread(target=func, args=(11,))
t1.start()
?
?
# 多线程方式：2
class MyThread(threading.Thread):
?
    def run(self):
        print(11111, self._args, self._kwargs)
        
t1 = MyThread(args=(11,))
t1.start()
?
t2 = MyThread(args=(22,))
t2.start()

运行结果：

11
11111 (11,) {}
11111 (22,) {}

3、Python 多线程

(1) 计算密集型多线程无用

import threading
?
v1 = [11,22,33] # +1
v2 = [44,55,66] # 100
?
def func(data,plus):
    for i in range(len(data)):
        data[i] = data[i] + plus
?
t1 = threading.Thread(target=func,args=(v1,1))
t1.start()
?
t2 = threading.Thread(target=func,args=(v2,100))
t2.start()

(2) IO操作 多线程有用

import threading
import requests
import uuid
?
url_list = [
    'http://pic.sc.chinaz.com/Files/pic/pic9/202008/apic27292_s.jpg',
    'http://pic.sc.chinaz.com/Files/pic/pic9/202008/apic27262_s.jpg',
    'http://pic1.sc.chinaz.com/Files/pic/pic9/202008/apic27250_s.jpg',
]
?
def task(url):
    ret = requests.get(url)
    file_name = str(uuid.uuid4()) + '.jpg'
    with open(file_name, mode='wb') as f:
        f.write(ret.content)
?
for url in url_list:
    t = threading.Thread(target=task, args=(url,))
    t.start()
 

(3) 多线程的问题

import time
import threading
lock = threading.RLock()
n = 10
?
def task(i):
    print('这段代码不加锁',i)
    lock.acquire() # 加锁，此区域的代码同一时刻只能有一个线程执行
    global n
    print('当前线程',i,'读取到的n值为：',n)
    n = i
    time.sleep(1)
    print('当前线程',i,'修改n值为：',n)
    lock.release() # 释放锁
?
for i in range(10):
    t = threading.Thread(target=task,args=(i,))
    t.start()

4、Python 线程锁

(1) 锁：Lock (1次放1个)

线程安全，多线程操作时，内部会让所有线程排队处理。如：list/dict/Queue

线程不安全 + 人(锁) => 排队处理。

"""
# 线程安全
import threading
v = []
def func(arg):
    v.append(arg)   # 线程安全
    print(v)
?
for i in range(10):
    t = threading.Thread(target=func, args=(i,))
    t.start()
"""
?
import threading
import time
?
v = []
lock = threading.Lock()
?
def func(arg):
    lock.acquire()  # 加锁
    v.append(arg)
    time.sleep(0.01)
    m = v[-1]
    print(arg, m)
    lock.release()  # 释放锁
?
for i in range(10):
    t = threading.Thread(target=func, args=(i,))
    t.start()

运行结果：

0 0
1 1
2 2
3 3
4 4
5 5
6 6
7 7
8 8
9 9

(2) 锁：RLock (1次放1个)

import threading
import time
?
v = []
lock = threading.RLock()
?
def func(arg):
    lock.acquire()
    lock.acquire()
?
    v.append(arg)
    time.sleep(0.01)
    m = v[-1]
    print(arg, m)
?
    lock.release()
    lock.release()
?
for i in range(10):
    t = threading.Thread(target=func, args=(i,))
    t.start()

 

(3) 锁：BoundedSemaphore（1次放N个）信号量

import threading
import time
?
v = []
lock = threading.BoundedSemaphore(3)
?
def func(arg):
    lock.acquire()
    print(arg)
    time.sleep(1)
    lock.release()
?
for i in range(20):
    t = threading.Thread(target=func, args=(i,))
    t.start()

运行结果：

0
1
2
3
4
5
6
7
8
9
10
11
12
13
14
16
15
17
18
19

(4) 锁：Condition（1次放指定个数）

方式一：

import time
import threading
?
lock = threading.Condition()
?
def func(arg):
    print('线程进来了')
    lock.acquire()
    lock.wait() # 加锁
?
    print(arg)
    time.sleep(1)
?
    lock.release()
?
for i in range(10):
    t =threading.Thread(target=func,args=(i,))
    t.start()
?
while True:
    inp = int(input('>>>'))
    lock.acquire()
    lock.notify(inp)
    lock.release()

方式二：

import time
import threading
?
lock = threading.Condition()
?
def xxxx():
    print('来执行函数了')
    input(">>>")
    # ct = threading.current_thread() # 获取当前线程
    # ct.getName()
    return True
?
def func(arg):
    print('线程进来了')
    lock.wait_for(xxxx)
    print(arg)
    time.sleep(1)
?
for i in range(10):
    t =threading.Thread(target=func,args=(i,))
    t.start()

(5) 锁：Event（1次放所有）

import time
import threading
?
lock = threading.Event()
def func(arg):
    print('线程来了')
    lock.wait()  # 加锁：红灯
    print(arg)
for i in range(10):
    t = threading.Thread(target=func, args=(i,))
    t.start()
input(">>>>")
lock.set()  # 绿灯
?
lock.clear()  # 再次变红灯
for i in range(10):
    t = threading.Thread(target=func, args=(i,))
    t.start()
input(">>>>")
lock.set()

5、threading.local()

内部自动为每个线程维护一个空间（字典），用于当前存取属于自己的值。保证线程之间的数据隔离。

{
    线程ID: {...}
    线程ID: {...}
    线程ID: {...}
    线程ID: {...}
}

(1) 基本使用

import time
import threading
?
v = threading.local()
?
def func(arg):
    # 内部会为当前线程创建一个空间用于存储：phone=自己的值
    v.phone = arg
    time.sleep(2)
    print(v.phone, arg)  # 去当前线程自己空间取值
?
for i in range(10):
    t = threading.Thread(target=func, args=(i,))
    t.start()

(2) 原理

import time
import threading
?
DATA_DICT = {}
?
def func(arg):
    # 获取线程的唯一标识
    ident = threading.get_ident()
    DATA_DICT[ident] = arg
    time.sleep(1)
    print(DATA_DICT[ident], arg)
?
for i in range(10):
    t = threading.Thread(target=func, args=(i,))
    t.start()

(3) 原理高级

import time
import threading
INFO = {}
class Local(object):
?
    def __getattr__(self, item):
        ident = threading.get_ident()
        return INFO[ident][item]
?
    def __setattr__(self, key, value):
        ident = threading.get_ident()
        if ident in INFO:
            INFO[ident][key] = value
        else:
            INFO[ident] = {key:value}
?
obj = Local()
?
def func(arg):
    obj.phone = arg # 调用对象的 __setattr__方法（“phone”,1）
    time.sleep(2)
    print(obj.phone,arg)
?
for i in range(10):
    t =threading.Thread(target=func,args=(i,))
    t.start()

6、线程池

from concurrent.futures import ThreadPoolExecutor
import time
?
def task(a1, a2):
    time.sleep(2)
    print(a1, a2)
?
# 创建了一个线程池（最多5个线程）
pool = ThreadPoolExecutor(5)
?
for i in range(40):
    # 去线程池中申请一个线程，让线程执行task函数。
    pool.submit(task, i, 8)

7、生成者消费者模型

三部件： 生产者 队列，先进先出 消费者 问：生产者消费者模型解决了什么问题？不用一直等待的问题。

import time
import queue
import threading
?
q = queue.Queue()  # 线程安全
?
def producer(id):
    """
    生产者
    :return:
    """
    while True:
        time.sleep(2)
        q.put('包子')
        print('厨师%s 生产了一个包子' % id)
?
for i in range(1, 4):
    t = threading.Thread(target=producer, args=(i,))
    t.start()
?
    
def consumer(id):
    """
    消费者
    :return:
    """
    while True:
        time.sleep(1)
        v1 = q.get()
        print('顾客 %s 吃了一个包子' % id)
?
for i in range(1, 3):
    t = threading.Thread(target=consumer, args=(i,))
    t.start()

 

第二章 Python 进程

1、Python 进程

(1) 初识进程

# windows
import multiprocessing
?
def task(arg):
    print(arg)
?
def run():
    for i in range(10):
        p = multiprocessing.Process(target=task, args=(i,))
        p.start()
?
if __name__ == '__main__':
    run()
    
# linux
"""
import multiprocessing
?
def task(arg):
    print(arg)
?
for i in range(10):
    p = multiprocessing.Process(target=task, args=(i,))
    p.start()
"""

运行结果：

1
0
2
4
3
5
6
7
8
9

(2) 进程间的数据不共享

import threading
import multiprocessing
?
data_list = []
?
def task(arg):
    data_list.append(arg)
    print(data_list)
?
def run():
    for i in range(10):
        p = multiprocessing.Process(target=task, args=(i,))
        # p = threading.Thread(target=task,args=(i,))
        p.start()
?
if __name__ == '__main__':
    run()

运行结果：

[0]
[8]
[3]
[2]
[1]
[6]
[5]
[7]
[9]
[4]

 

(3) 进程常用功能

import time
import multiprocessing
?
def task(arg):
    p = multiprocessing.current_process()
    print(p.name)   # 获取进程名称
    print(p.ident, p.pid)   # 获取进程id
    time.sleep(2)
    print(arg)
?
?
def run():
    print('111111111')
    p1 = multiprocessing.Process(target=task, args=(1,))
    p1.daemon = True
    p1.name = 'pp1'  # 设置进程名称
    p1.start()  # 启动进程
    p1.join()  # 等待进程
    print('222222222')
?
    p2 = multiprocessing.Process(target=task, args=(2,))
    p2.daemon = True
    p2.name = 'pp2'  # 设置进程名称
    p2.start()  # 启动进程
    p2.join()  # 等待进程
    print('333333333')
?
if __name__ == '__main__':
    run()

(4) 通过继承方式创建进程

import multiprocessing
?
class MyProcess(multiprocessing.Process):
    def run(self):
        print('当前进程', multiprocessing.current_process())
?
def run():
    p1 = MyProcess()
    p1.start()
    
    p2 = MyProcess()
    p2.start()
?
if __name__ == '__main__':
    run()

2、Python 进程数据共享

(1) Queue

# windows
import time
import multiprocessing
?
def task(arg, q):
    q.put(arg)
?
def run():
    q = multiprocessing.Queue()
    for i in range(10):
        p = multiprocessing.Process(target=task, args=(i, q,))
        p.start()
?
    time.sleep(2)
    while True:
        v = q.get()
        print(v)
?
if __name__ == '__main__':
    run()
    
# linux
"""
import multiprocessing
?
q = multiprocessing.Queue()
?
def task(arg, q):
    q.put(arg)
?
def run():
    for i in range(10):
        p = multiprocessing.Process(target=task, args=(i, q,))
        p.start()
?
    while True:
        v = q.get()
        print(v)
"""

运行结果：

8
3
2
0
1
5
6
7
4
9

(2) Manager

# windows
import time
import multiprocessing
?
def task(arg, dic):
    time.sleep(2)
    dic[arg] = 100
?
if __name__ == '__main__':
    m = multiprocessing.Manager()
    dic = m.dict()
    process_list = []
    for i in range(10):
        p = multiprocessing.Process(target=task, args=(i, dic,))
        p.start()
        # p.join() ***
        process_list.append(p)
?
    while True:
        count = 0
        for p in process_list:
            if not p.is_alive():
                count += 1
        if count == len(process_list):
            break
    print(dic)
    
# linux
"""
import multiprocessing
m = multiprocessing.Manager()
dic = m.dict()
?
def task(arg):
    dic[arg] = 100
?
def run():
    for i in range(10):
        p = multiprocessing.Process(target=task, args=(i,))
        p.start()
?
    input('>>>')
    print(dic.values())
?
if __name__ == '__main__':
    run()
"""

运行结果：

{0: 100, 1: 100, 2: 100, 3: 100, 4: 100, 5: 100, 6: 100, 7: 100, 8: 100, 9: 100}

3、Python 进程锁

import time
import threading
import multiprocessing
?
lock = multiprocessing.RLock()
?
def task(arg):
    print('鬼子来了')
    lock.acquire()
    time.sleep(2)
    print(arg)
    lock.release()
?
?
if __name__ == '__main__':
    p1 = multiprocessing.Process(target=task, args=(1,))
    p1.start()
?
    p2 = multiprocessing.Process(target=task, args=(2,))
    p2.start()

运行结果：

鬼子来了
鬼子来了
1
2

4、Python 进程池

import time
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
?
def task(arg):
    time.sleep(2)
    print(arg)
?
if __name__ == '__main__':
    pool = ProcessPoolExecutor(5)
    for i in range(10):
        pool.submit(task, i)

运行结果:

0
1
2
3
4
5
6
7
8
9

 

第三章 Python 协程

概念：

进程，操作系统中存在；

线程，操作系统中存在；

协程，是由程序员创造出来的一个不是真实存在的东西；

协程：是微线程，对一个线程进程分片，使得线程在代码块之间进行来回切换执行，而不是在原来逐行执行。

1、greenlet

greentlet是一个第三方模块，需要提前安装 pip install greenlet才能使用。

from greenlet import greenlet
def func1():
    print(1)        # 第1步：输出 1
    gr2.switch()    # 第3步：切换到 func2 函数
    print(2)        # 第6步：输出 2
    gr2.switch()    # 第7步：切换到 func2 函数，从上一次执行的位置继续向后执行
def func2():
    print(3)        # 第4步：输出 3
    gr1.switch()    # 第5步：切换到 func1 函数，从上一次执行的位置继续向后执行
    print(4)        # 第8步：输出 4
gr1 = greenlet(func1)
gr2 = greenlet(func2)
gr1.switch() # 第1步：去执行 func1 函数

运行结果:

1
3
2
4

注意：单纯的协程无用

注意：switch中也可以传递参数用于在切换执行时相互传递值。

 

2、yield

基于Python的生成器的yield和yield form关键字实现协程代码。

def func1():
    yield 1
    yield from func2()
    yield 2
def func2():
    yield 3
    yield 4
f1 = func1()
for item in f1:
    print(item)

运行结果:

1
3
4
2

注意：yield form关键字是在Python3.3中引入的。

 

3、gevent

协程 + 遇到IO就切换 => 牛逼起来了 pip install gevent

from gevent import monkey
?
monkey.patch_all()  # 以后代码中遇到IO都会自动执行greenlet的switch进行切换
import requests
import gevent
?
def get_page1(url):
    ret = requests.get(url)
    print(url, ret.content)
?
def get_page2(url):
    ret = requests.get(url)
    print(url, ret.content)
?
def get_page3(url):
    ret = requests.get(url)
    print(url, ret.content)
?
gevent.joinall([
    gevent.spawn(get_page1, 'https://www.python.org/'),  # 协程1
    gevent.spawn(get_page2, 'https://www.yahoo.com/'),  # 协程2
    gevent.spawn(get_page3, 'https://github.com/'),  # 协程3
])

4、总结

1. 什么是协程？ 协程也可以称为“微线程”，就是开发者控制线程执行流程，控制先执行某段代码然后再切换到另外函执行代码...来回切换。

2. 协程可以提高并发吗？ 协程自己本身无法实现并发（甚至性能会降低）。 协程+IO切换性能提高。

3. 进程、线程、协程的区别？

4. 单线程提供并发：

   协程+IO切换：gevent

   基于事件循环的异步非阻塞框架：Twisted

 

 

第四章 Python 异步

1、前言

想学asyncio，得先了解协程，协程是根本呀！

协程（Coroutine），也可以被称为微线程，是一种用户态内的上下文切换技术。简而言之，其实就是通过一个线程实现代码块相互切换执行。例如：

def func1():
    print(1)
    ...
    print(2)
def func2():
    print(3)
    ...
    print(4)
func1()
func2()

上述代码是普通的函数定义和执行，按流程分别执行两个函数中的代码，并先后会输出：1、2、3、4。但如果介入协程技术那么就可以实现函数见代码切换执行，最终输入：1、3、2、4 。

在Python中有多种方式可以实现协程，例如：

• greenlet，是一个第三方模块，用于实现协程代码（Gevent协程就是基于greenlet实现）

• yield，生成器，借助生成器的特点也可以实现协程代码。

• asyncio，在Python3.4中引入的模块用于编写协程代码。

• async & awiat，在Python3.5中引入的两个关键字，结合asyncio模块可以更方便的编写协程代码。

 

2、asyncio

(1) 基本使用

在Python3.4之前官方未提供协程的类库，一般大家都是使用greenlet等其他来实现。在Python3.4发布后官方正式支持协程，即：asyncio模块。

import asyncio
?
@asyncio.coroutine
def func1():
    print(1)
    yield from asyncio.sleep(2)  # 遇到IO耗时操作，自动化切换到tasks中的其他任务
    print(2)
?
@asyncio.coroutine
def func2():
    print(3)
    yield from asyncio.sleep(2)  # 遇到IO耗时操作，自动化切换到tasks中的其他任务
    print(4)
?
tasks = [
    asyncio.ensure_future(func1()),
    asyncio.ensure_future(func2())
]
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(tasks))

运行结果

1
3
2
4

注意：基于asyncio模块实现的协程比之前的要更厉害，因为他的内部还集成了遇到IO耗时操作自动切花的功能。

 

(2) async & awit

async & awit 关键字在Python3.5版本中正式引入，基于他编写的协程代码其实就是 上一示例 的加强版，让代码可以更加简便。

Python3.8之后 @asyncio.coroutine 装饰器就会被移除，推荐使用async & awit 关键字实现协程代码。

import asyncio
?
async def func1():
    print(1)
    await asyncio.sleep(2)
    print(2)
?
?
async def func2():
    print(3)
    await asyncio.sleep(2)
    print(4)
?
?
tasks = [
    asyncio.ensure_future(func1()),
    asyncio.ensure_future(func2())
]
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(tasks))

运行结果

1
3
2
4

(3) 小结

关于协程有多种实现方式，目前主流使用是Python官方推荐的asyncio模块和async&await关键字的方式，例如：在tonado、sanic、fastapi、django3 中均已支持。

接下来，我们也会针对 asyncio模块 + async & await 关键字进行更加详细的讲解。

 

3、协程的意义

通过学习，我们已经了解到协程可以通过一个线程在多个上下文中进行来回切换执行。

但是，协程来回切换执行的意义何在呢？（网上看到很多文章舔协程，协程牛逼之处是哪里呢？）

1. 计算型的操作，利用协程来回切换执行，没有任何意义，来回切换并保存状态 反倒会降低性能。

2. IO型的操作，利用协程在IO等待时间就去切换执行其他任务，当IO操作结束后再自动回调，那么就会大大节省资源并提供性能，从而实现异步编程（不等待任务结束就可以去执行其他代码）。

(1) 爬虫案例

例如：用代码实现下载 url_list 中的图片。

1-1 同步编程实现

"""
下载图片使用第三方模块requests，请提前安装：pip3 install requests
"""
import requests
?
def download_image(url):
    print("开始下载:", url)
    # 发送网络请求，下载图片
    response = requests.get(url)
    print("下载完成")
    # 图片保存到本地文件
    file_name = url.rsplit('_')[-1]
    with open(file_name, mode='wb') as file_object:
        file_object.write(response.content)
?
if __name__ == '__main__':
    url_list = [
        'https://www3.autoimg.cn/newsdfs/g26/M02/35/A9/120x90_0_autohomecar__ChsEe12AXQ6AOOH_AAFocMs8nzU621.jpg',
        'https://www2.autoimg.cn/newsdfs/g30/M01/3C/E2/120x90_0_autohomecar__ChcCSV2BBICAUntfAADjJFd6800429.jpg',
        'https://www3.autoimg.cn/newsdfs/g26/M0B/3C/65/120x90_0_autohomecar__ChcCP12BFCmAIO83AAGq7vK0sGY193.jpg'
    ]
    for item in url_list:
        download_image(item)

1-2 基于协程的异步编程实现

"""
下载图片使用第三方模块aiohttp，请提前安装：pip3 install aiohttp
"""
# !/usr/bin/env python
# -*- coding:utf-8 -*-
import aiohttp
import asyncio
?
async def fetch(session, url):
    print("发送请求：", url)
    async with session.get(url, verify_ssl=False) as response:
        content = await response.content.read()
        file_name = url.rsplit('_')[-1]
        with open(file_name, mode='wb') as file_object:
            file_object.write(content)
?
async def main():
    async with aiohttp.ClientSession() as session:
        url_list = [
            'https://www3.autoimg.cn/newsdfs/g26/M02/35/A9/120x90_0_autohomecar__ChsEe12AXQ6AOOH_AAFocMs8nzU621.jpg',
            'https://www2.autoimg.cn/newsdfs/g30/M01/3C/E2/120x90_0_autohomecar__ChcCSV2BBICAUntfAADjJFd6800429.jpg',
            'https://www3.autoimg.cn/newsdfs/g26/M0B/3C/65/120x90_0_autohomecar__ChcCP12BFCmAIO83AAGq7vK0sGY193.jpg'
        ]
        tasks = [asyncio.create_task(fetch(session, url)) for url in url_list]
        await asyncio.wait(tasks)
?
if __name__ == '__main__':
    asyncio.run(main())

上述两种的执行对比之后会发现，基于协程的异步编程 要比 同步编程的效率高了很多。因为：

• 同步编程，按照顺序逐一排队执行，如果图片下载时间为2分钟，那么全部执行完则需要6分钟。

• 异步编程，几乎同时发出了3个下载任务的请求（遇到IO请求自动切换去发送其他任务请求），如果图片下载时间为2分钟，那么全部执行完毕也大概需要2分钟左右就可以了。

 

(2) 小结

协程一般应用在有IO操作的程序中，因为协程可以利用IO等待的时间去执行一些其他的代码，从而提升代码执行效率。

生活中不也是这样的么，假设 你是一家制造汽车的老板，员工点击设备的【开始】按钮之后，在设备前需等待30分钟，然后点击【结束】按钮，此时作为老板的你一定希望这个员工在等待的那30分钟的时间去做点其他的工作。

 

4、异步编程

基于async & await关键字的协程可以实现异步编程，这也是目前python异步相关的主流技术。

想要真正的了解Python中内置的异步编程，根据下文的顺序一点点来看。

(1) 事件循环

事件循环，可以把他当做是一个while循环，这个while循环在周期性的运行并执行一些任务，在特定条件下终止循环。

# 伪代码
任务列表 = [ 任务1, 任务2, 任务3,... ]
while True:
    可执行的任务列表，已完成的任务列表 = 去任务列表中检查所有的任务，将'可执行'和'已完成'的任务返回
    for 就绪任务 in 已准备就绪的任务列表:
        执行已就绪的任务
    for 已完成的任务 in 已完成的任务列表:
        在任务列表中移除 已完成的任务
    如果 任务列表 中的任务都已完成，则终止循环

在编写程序时候可以通过如下代码来获取和创建事件循环。

import asyncio
loop = asyncio.get_event_loop()

(2) 协程和异步编程

协程函数，定义形式为 async def 的函数。

协程对象，调用 协程函数 所返回的对象

# 定义一个协程函数
async def func():
    pass
?
# 调用协程函数，返回一个协程对象
result = func()

注意：调用协程函数时，函数内部代码不会执行，只是会返回一个协程对象。

 

2-1 基本应用

程序中，如果想要执行协程函数的内部代码，需要 事件循环 和 协程对象 配合才能实现，如：

import asyncio
async def func():
    print("协程内部代码")
?
# 调用协程函数，返回一个协程对象。
result = func()
"""
# python 3.7 以下版本
# 创建一个事件循环
# 将协程当做任务提交到事件循环的任务列表中，协程执行完成之后终止。
"""
loop = asyncio.get_event_loop()
loop.run_until_complete(result)
?
"""
# python 3.7 及以上版本
本质上方式一是一样的，内部先 创建事件循环 然后执行 run_until_complete，一个简便的写法。
asyncio.run 函数在 Python 3.7 中加入 asyncio 模块，
asyncio.run(result)
"""

这个过程可以简单理解为：将协程当做任务添加到 事件循环 的任务列表，然后事件循环检测列表中的协程是否 已准备就绪（默认可理解为就绪状态），如果准备就绪则执行其内部代码。

 

2-2 await

await是一个只能在协程函数中使用的关键字，用于遇到IO操作时挂起 当前协程（任务），当前协程（任务）挂起过程中 事件循环可以去执行其他的协程（任务），当前协程IO处理完成时，可以再次切换回来执行await之后的代码。代码如下：

案例1

import asyncio
?
async def func():
    print("执行协程函数内部代码")
    # 遇到IO操作挂起当前协程（任务），等IO操作完成之后再继续往下执行。
    # 当前协程挂起时，事件循环可以去执行其他协程（任务）。
    response = await asyncio.sleep(2)
    print("IO请求结束，结果为：", response)
?
result = func()
loop = asyncio.get_event_loop()
loop.run_until_complete(result)

案例2

import asyncio
?
async def others():
    print("start")
    await asyncio.sleep(2)
    print('end')
    return '返回值'
?
async def func():
    print("执行协程函数内部代码")
    # 遇到IO操作挂起当前协程（任务），等IO操作完成之后再继续往下执行。当前协程挂起时，事件循环可以去执行其他协程（任务）。
    response = await others()
    print("IO请求结束，结果为：", response)
?
loop = asyncio.get_event_loop()
loop.run_until_complete(func())

案例3

import asyncio
?
async def others():
    print("start")
    await asyncio.sleep(2)
    print('end')
    return '返回值'
?
async def func():
    print("执行协程函数内部代码")
    # 遇到IO操作挂起当前协程（任务），等IO操作完成之后再继续往下执行。当前协程挂起时，事件循环可以去执行其他协程（任务）。
    response1 = await others()
    print("IO请求结束，结果为：", response1)
    response2 = await others()
    print("IO请求结束，结果为：", response2)
?
loop = asyncio.get_event_loop()
loop.run_until_complete(func())

上述的所有示例都只是创建了一个任务，即：事件循环的任务列表中只有一个任务，所以在IO等待时无法演示切换到其他任务效果。

在程序想要创建多个任务对象，需要使用Task对象来实现。

 

2-3 Task对象

https://docs.python.org/3.8/library/asyncio-task.html#asyncio.create_task

Tasks用于并发调度协程，通过asyncio.create_task(协程对象)的方式创建Task对象，这样可以让协程加入事件循环中等待被调度执行。除了使用 asyncio.create_task() 函数以外，还可以用低层级的 loop.create_task() 或 ensure_future() 函数。不建议手动实例化 Task 对象。

本质上是将协程对象封装成task对象，并将协程立即加入事件循环，同时追踪协程的状态。

注意：asyncio.create_task() 函数在 Python 3.7 中被加入。在 Python 3.7 之前，可以改用低层级的 asyncio.ensure_future() 函数。

• 案例1

• import asyncio
  ?
  async def func():
      print(1)
      await asyncio.sleep(2)
      print(2)
      return "返回值"
  ?
  async def main():
      print("main开始")
      
      loop = asyncio.get_event_loop()
      # 创建协程，将协程封装到一个Task对象中并立即添加到事件循环的任务列表中，等待事件循环去执行（默认是就绪状态）。
      task1 = loop.create_task(func())
      # 创建协程，将协程封装到一个Task对象中并立即添加到事件循环的任务列表中，等待事件循环去执行（默认是就绪状态）。
      task2 = loop.create_task(func())
      print("main结束")
      # 当执行某协程遇到IO操作时，会自动化切换执行其他任务。
      # 此处的await是等待相对应的协程全都执行完毕并获取结果
      ret1 = await task1
      ret2 = await task2
      print(ret1, ret2)
  ?
  loop = asyncio.get_event_loop()
  loop.run_until_complete(main())

• 案例2

• import asyncio
  ?
  async def func():
      print(1)
      await asyncio.sleep(2)
      print(2)
      return "返回值"
  ?
  async def main():
      print("main开始")
      # 创建协程，将协程封装到Task对象中并添加到事件循环的任务列表中，等待事件循环去执行（默认是就绪状态）。
      # 在调用
      loop = asyncio.get_event_loop()
      task_list = [
          loop.create_task(func()),
          loop.create_task(func())
      ]
      print("main结束")
      # 当执行某协程遇到IO操作时，会自动化切换执行其他任务。
      # 此处的await是等待所有协程执行完毕，并将所有协程的返回值保存到done
      # 如果设置了timeout值，则意味着此处最多等待的秒，完成的协程返回值写入到done中，未完成则写到pending中。
      done, pending = await asyncio.wait(task_list, timeout=None)
      print(done, pending)
  ?
  loop = asyncio.get_event_loop()
  loop.run_until_complete(main())
  

•  

  注意：asyncio.wait 源码内部会对列表中的每个协程执行ensure_future从而封装为Task对象，所以在和wait配合使用时task_list的值为[func(),func()] 也是可以的。

• 案例3

• import asyncio
  ?
  async def func():
      print("执行协程函数内部代码")
      # 遇到IO操作挂起当前协程（任务），等IO操作完成之后再继续往下执行。当前协程挂起时，事件循环可以去执行其他协程（任务）。
      response = await asyncio.sleep(2)
      print("IO请求结束，结果为：", response)
  ?
  ?
  coroutine_list = [func(), func()]
  # 错误：coroutine_list = [ asyncio.create_task(func()), asyncio.create_task(func()) ]
  # 此处不能直接 asyncio.create_task，因为将Task立即加入到事件循环的任务列表，
  # 但此时事件循环还未创建，所以会报错。
  # 使用asyncio.wait将列表封装为一个协程，并调用asyncio.run实现执行两个协程
  # asyncio.wait内部会对列表中的每个协程执行ensure_future，封装为Task对象。
  loop = asyncio.get_event_loop()
  done, pending = loop.run_until_complete(asyncio.wait(coroutine_list))
  

2-4 asyncio.Future对象

asyncio中的Future对象是一个相对更偏向底层的可对象，通常我们不会直接用到这个对象，而是直接使用Task对象来完成任务的并和状态的追踪。（ Task 是 Futrue的子类 ）

Future为我们提供了异步编程中的 最终结果 的处理（Task类也具备状态处理的功能）。

• 案例1

• import asyncio
  ?
  async def main():
      # 获取当前事件循环
      loop = asyncio.get_event_loop()
      # # 创建一个任务（Future对象），这个任务什么都不干。
      fut = loop.create_future()
      # 等待任务最终结果（Future对象），没有结果则会一直等下去。
      await fut
  ?
  loop = asyncio.get_event_loop()
  loop.run_until_complete(main())
  

• 案例2

• import asyncio
  ?
  async def set_after(fut):
      await asyncio.sleep(2)
      fut.set_result("666")
  ?
  async def main():
      # 获取当前事件循环
      loop = asyncio.get_event_loop()
      # 创建一个任务（Future对象），没绑定任何行为，则这个任务永远不知道什么时候结束。
      fut = loop.create_future()
      # 创建一个任务（Task对象），绑定了set_after函数，函数内部在2s之后，会给fut赋值。
      # 即手动设置future任务的最终结果，那么fut就可以结束了。
      await loop.create_task(set_after(fut))
      # 等待 Future对象获取 最终结果，否则一直等下去
      data = await fut
      print(data)
  ?
  loop = asyncio.get_event_loop()
  loop.run_until_complete(main())

• Future对象本身函数进行绑定，所以想要让事件循环获取Future的结果，则需要手动设置。而Task对象继承了Future对象，其实就对Future进行扩展，他可以实现在对应绑定的函数执行完成之后，自动执行set_result，从而实现自动结束。

  虽然，平时使用的是Task对象，但对于结果的处理本质是基于Future对象来实现的。

  扩展：支持 await 对象语 法的对象课成为可等待对象，所以 协程对象、Task对象、Future对象 都可以被成为可等待对象。

 

2-5 futures.Future对象

在Python的concurrent.futures模块中也有一个Future对象，这个对象是基于线程池和进程池实现异步操作时使用的对象。

import time
from concurrent.futures import Future
from concurrent.futures.thread import ThreadPoolExecutor
from concurrent.futures.process import ProcessPoolExecutor
?
def func(value):
    time.sleep(1)
    print(value)
?
pool = ThreadPoolExecutor(max_workers=5)
# 或 pool = ProcessPoolExecutor(max_workers=5)
for i in range(10):
    fut = pool.submit(func, i)
    print(fut)

两个Future对象是不同的，他们是为不同的应用场景而设计，例如：concurrent.futures.Future不支持await语法 等。

官方提示两对象之间不同：

• unlike asyncio Futures, concurrent.futures.Future instances cannot be awaited.

• asyncio.Future.result() and asyncio.Future.exception() do not accept the timeout argument.

• asyncio.Future.result() and asyncio.Future.exception() raise an InvalidStateError exception when the Future is not done.

• Callbacks registered with asyncio.Future.add_done_callback() are not called immediately. They are scheduled with loop.call_soon() instead.

• asyncio Future is not compatible with the concurrent.futures.wait() and concurrent.futures.as_completed() functions.

在Python提供了一个将futures.Future 对象包装成asyncio.Future对象的函数 asynic.wrap_future。

接下里你肯定问：为什么python会提供这种功能？

其实，一般在程序开发中我们要么统一使用 asycio 的协程实现异步操作、要么都使用进程池和线程池实现异步操作。但如果 协程的异步和 进程池/线程池的异步 混搭时，那么就会用到此功能了

import time
import asyncio
import concurrent.futures
?
def func1():
    # 某个耗时操作
    time.sleep(2)
    return "SB"
?
async def main():
    loop = asyncio.get_event_loop()
    # 1. Run in the default loop's executor ( 默认ThreadPoolExecutor )
    # 第一步：内部会先调用 ThreadPoolExecutor 的 submit 方法去线程池中申请一个线程去执行func1函数，并返回一个concurrent.futures.Future对象
    # 第二步：调用asyncio.wrap_future将concurrent.futures.Future对象包装为asycio.Future对象。
    # 因为concurrent.futures.Future对象不支持await语法，所以需要包装为 asycio.Future对象 才能使用。
    fut = loop.run_in_executor(None, func1)
    result = await fut
    print('default thread pool', result)
?
    # 2. Run in a custom thread pool:
    # with concurrent.futures.ThreadPoolExecutor() as pool:
    #     result = await loop.run_in_executor(
    #         pool, func1)
    #     print('custom thread pool', result)
    # 3. Run in a custom process pool:
    # with concurrent.futures.ProcessPoolExecutor() as pool:
    #     result = await loop.run_in_executor(
    #         pool, func1)
    #     print('custom process pool', result)
?
loop = asyncio.get_event_loop()
loop.run_until_complete(main())

应用场景：当项目以协程式的异步编程开发时，如果要使用一个第三方模块，而第三方模块不支持协程方式异步编程时，就需要用到这个功能，例如：

import asyncio
import requests
?
async def download_image(url):
    # 发送网络请求，下载图片（遇到网络下载图片的IO请求，自动化切换到其他任务）
    print("开始下载:", url)
    loop = asyncio.get_event_loop()
    # requests模块默认不支持异步操作，所以就使用线程池来配合实现了。
    future = loop.run_in_executor(None, requests.get, url)
    response = await future
    print('下载完成')
    # 图片保存到本地文件
    file_name = url.rsplit('/')[-1]
    with open(file_name, mode='wb') as file_object:
        file_object.write(response.content)
?
if __name__ == '__main__':
    url_list = [
        'http://pic.sc.chinaz.com/Files/pic/pic9/202008/hpic2824_s.jpg',
        'http://pic.sc.chinaz.com/Files/pic/pic9/202008/hpic2825_s.jpg',
        'http://pic.sc.chinaz.com/Files/pic/pic9/202008/hpic2828_s.jpg'
    ]
    tasks = [download_image(url) for url in url_list]
    loop = asyncio.get_event_loop()
    loop.run_until_complete(asyncio.wait(tasks))

2-6 异步迭代器

• 什么是异步迭代器

  实现了 __aiter__() 和 __anext__() 方法的对象。__anext__ 必须返回一个 awaitable 对象。async for 会处理异步迭代器的 __anext__() 方法所返回的可等待对象，直到其引发一个 StopAsyncIteration 异常。由 PEP 492 引入。

• 什么是异步可迭代对象？

  可在 async for 语句中被使用的对象。必须通过它的 __aiter__() 方法返回一个 asynchronous iterator。由 PEP 492 引入。

• 基本使用

• import asyncio
  ?
  class Reader(object):
      """ 自定义异步迭代器（同时也是异步可迭代对象） """
  ?
      def __init__(self):
          self.count = 0
  ?
      async def readline(self):
          # await asyncio.sleep(1)
          self.count += 1
          if self.count == 100:
              return None
          return self.count
  ?
      def __aiter__(self):
          return self
  ?
      async def __anext__(self):
          val = await self.readline()
          if val == None:
              raise StopAsyncIteration
          return val
  ?
  async def func():
      # 创建异步可迭代对象
      async_iter = Reader()
      # async for 必须要放在async def函数内，否则语法错误。
      async for item in async_iter:
          print(item)
  ?
  loop = asyncio.get_event_loop()
  loop.run_until_complete(func())

• 异步迭代器其实没什么太大的作用，只是支持了async for语法而已。

 

2-7 异步上下文管理器

此种对象通过定义 __aenter__() 和 __aexit__() 方法来对 async with 语句中的环境进行控制。由 PEP 492 引入。

import asyncio
?
class AsyncContextManager:
    def __init__(self):
        self.conn = None
?
    async def do_something(self):
        # 异步操作数据库
        return 666
?
    async def __aenter__(self):
        # 异步链接数据库
        self.conn = await asyncio.sleep(1)
        return self
?
    async def __aexit__(self, exc_type, exc, tb):
        # 异步关闭数据库链接
        await asyncio.sleep(1)
?
async def func():
    async with AsyncContextManager() as f:
        result = await f.do_something()
        print(result)
?
loop = asyncio.get_event_loop()
loop.run_until_complete(func())

这个异步的上下文管理器还是比较有用的，平时在开发过程中 打开、处理、关闭 操作时，就可以用这种方式来处理。

 

(3) 小结

在程序中只要看到async和await关键字，其内部就是基于协程实现的异步编程，这种异步编程是通过一个线程在IO等待时间去执行其他任务，从而实现并发。

以上就是异步编程的常见操作，内容参考官方文档。

• 中文版：https://docs.python.org/zh-cn/3.8/library/asyncio.html

• 英文本：https://docs.python.org/3.8/library/asyncio.html

5、uvloop

Python标准库中提供了asyncio模块，用于支持基于协程的异步编程。

uvloop是 asyncio 中的事件循环的替代方案，替换后可以使得asyncio性能提高。事实上，uvloop要比nodejs、gevent等其他python异步框架至少要快2倍，性能可以比肩Go语言。

(1) 安装uvloop

pip3 install uvloop

 

(2) 基本使用

在项目中想要使用uvloop替换asyncio的事件循环也非常简单，只要在代码中这么做就行。

import asyncio
import uvloop
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
# 编写asyncio的代码，与之前写的代码一致。
# 内部的事件循环自动化会变为uvloop
asyncio.run(...)

注意：知名的asgi uvicorn内部就是使用的uvloop的事件循环。

 

6、实战案例

(1) 异步Redis

当通过python去操作redis时，链接、设置值、获取值 这些都涉及网络IO请求，使用asycio异步的方式可以在IO等待时去做一些其他任务，从而提升性能。

安装Python异步操作redis模块

pip3 install aioredis

1-1 案例：异步操作Redis

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import asyncio
import aioredis
?
async def execute(address, password):
    print("开始执行", address)
    # 网络IO操作：创建redis连接
    redis = await aioredis.create_redis(address, password=password)
    # 网络IO操作：在redis中设置哈希值car，内部在设三个键值对，即： redis = { car:{key1:1,key2:2,key3:3}}
    await redis.hmset_dict('car', key1=1, key2=2, key3=3)
    # 网络IO操作：去redis中获取值
    result = await redis.hgetall('car', encoding='utf-8')
    print(result)
    redis.close()
    # 网络IO操作：关闭redis连接
    await redis.wait_closed()
    print("结束", address)
?
loop = asyncio.get_event_loop()
loop.run_until_complete(execute('redis://127.0.0.1:6379', "root!123456"))

1-2 案例：连接多个Redis做操作

遇到IO会切换其他任务，提供了性能。

import asyncio
import aioredis
?
async def execute(address, password):
    print("开始执行", address)
    # 网络IO操作：先去连接 47.93.4.197:6379，遇到IO则自动切换任务，去连接47.93.4.198:6379
    redis = await aioredis.create_redis_pool(address, password=password)
    # 网络IO操作：遇到IO会自动切换任务
    await redis.hmset_dict('car', key1=1, key2=2, key3=3)
    # 网络IO操作：遇到IO会自动切换任务
    result = await redis.hgetall('car', encoding='utf-8')
    print(result)
    redis.close()
    # 网络IO操作：遇到IO会自动切换任务
    await redis.wait_closed()
    print("结束", address)
?
task_list = [
    execute('redis://47.93.4.197:6379', "root!2345"),
    execute('redis://47.93.4.198:6379', "root!2345")
]
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(task_list))

更多redis操作参考aioredis官网：https://aioredis.readthedocs.io/en/v1.3.0/start.html

 

(2) 异步MySQL

当通过python去操作MySQL时，连接、执行SQL、关闭都涉及网络IO请求，使用asycio异步的方式可以在IO等待时去做一些其他任务，从而提升性能。

安装Python异步操作redis模块

pip3 install aiomysql

2-1 案例：异步操作MySQL

import asyncio
import aiomysql
?
async def execute():
    # 网络IO操作：连接MySQL
    conn = await aiomysql.connect(host='127.0.0.1', port=3306, user='root', password='19971215', db='mysql', )
    # 网络IO操作：创建CURSOR
    cur = await conn.cursor()
    # 网络IO操作：执行SQL
    await cur.execute("SELECT Host,User FROM user")
    # 网络IO操作：获取SQL结果
    result = await cur.fetchall()
    print(result)
    # 网络IO操作：关闭链接
    await cur.close()
    conn.close()
?
loop = asyncio.get_event_loop()
loop.run_until_complete(execute())

2-2 案例：连接多个MySQL做操作

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import asyncio
import aiomysql
?
async def execute(host, password):
    print("开始", host)
    # 网络IO操作：先去连接 47.93.40.197，遇到IO则自动切换任务，去连接47.93.40.198:6379
    conn = await aiomysql.connect(host=host, port=3306, user='root', password=password, db='mysql')
    # 网络IO操作：遇到IO会自动切换任务
    cur = await conn.cursor()
    # 网络IO操作：遇到IO会自动切换任务
    await cur.execute("SELECT Host,User FROM user")
    # 网络IO操作：遇到IO会自动切换任务
    result = await cur.fetchall()
    print(result)
    # 网络IO操作：遇到IO会自动切换任务
    await cur.close()
    conn.close()
    print("结束", host)
?
task_list = [
    execute('127.0.0.1', "19971215"),
    execute('47.108.63.228', "19971215")
]
?
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(task_list))

(3) FastAPI框架

FastAPI是一款用于构建API的高性能web框架，框架基于Python3.6+的 type hints搭建。

接下里的异步示例以FastAPI和uvicorn来讲解（uvicorn是一个支持异步的asgi）。

安装FastAPI web 框架，

pip3 install fastapi

安装uvicorn，本质上为web提供socket server的支持的asgi（一般支持异步称asgi、不支持异步称wsgi）

pip3 install uvicorn

3-1 案例

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import asyncio
import uvicorn
import aioredis
from aioredis import Redis
from fastapi import FastAPI
?
app = FastAPI()
REDIS_POOL = aioredis.ConnectionsPool('redis://47.193.14.198:6379', password="root123", minsize=1, maxsize=10)
?
@app.get("/")
def index():
    """ 普通操作接口 """
    return {"message": "Hello World"}
?
@app.get("/red")
async def red():
    """ 异步操作接口 """
    print("请求来了")
    await asyncio.sleep(3)
    # 连接池获取一个连接
    conn = await REDIS_POOL.acquire()
    redis = Redis(conn)
    # 设置值
    await redis.hmset_dict('car', key1=1, key2=2, key3=3)
    # 读取值
    result = await redis.hgetall('car', encoding='utf-8')
    print(result)
    # 连接归还连接池
    REDIS_POOL.release(conn)
    return result
?
if __name__ == '__main__':
    uvicorn.run("luffy:app", host="127.0.0.1", port=5000, log_level="info")

在有多个用户并发请求的情况下，异步方式来编写的接口可以在IO等待过程中去处理其他的请求，提供性能。

例如：同时有两个用户并发来向接口 http://127.0.0.1:5000/red 发送请求，服务端只有一个线程，同一时刻只有一个请求被处理。 异步处理可以提供并发是因为：当视图函数在处理第一个请求时，第二个请求此时是等待被处理的状态，当第一个请求遇到IO等待时，会自动切换去接收并处理第二个请求，当遇到IO时自动化切换至其他请求，一旦有请求IO执行完毕，则会再次回到指定请求向下继续执行其功能代码。

基于上下文管理，来实现自动化管理的案例：

 

3-2 案例：Redis

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import asyncio
import uvicorn
import aioredis
from aioredis import Redis
from fastapi import FastAPI
?
app = FastAPI()
REDIS_POOL = aioredis.ConnectionsPool('redis://47.193.14.198:6379', password="root123", minsize=1, maxsize=10)
?
@app.get("/")
def index():
    """ 普通操作接口 """
    return {"message": "Hello World"}
?
@app.get("/red")
async def red():
    """ 异步操作接口 """
    print("请求来了")
    async with REDIS_POOL.get() as conn:
        redis = Redis(conn)
        # 设置值
        await redis.hmset_dict('car', key1=1, key2=2, key3=3)
        # 读取值
        result = await redis.hgetall('car', encoding='utf-8')
        print(result)
    return result
?
if __name__ == '__main__':
    uvicorn.run("fast3:app", host="127.0.0.1", port=5000, log_level="info")

3-3 案例：MySQL

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import asyncio
import uvicorn
from fastapi import FastAPI
import aiomysql
?
app = FastAPI()
# 创建数据库连接池
pool = aiomysql.Pool(host='127.0.0.1', port=3306, user='root', password='123', db='mysql',
                     minsize=1, maxsize=10, echo=False, pool_recycle=-1, loop=asyncio.get_event_loop())
?
@app.get("/red")
async def red():
    """ 异步操作接口 """
    # 去数据库连接池申请链接
    async with pool.acquire() as conn:
        async with conn.cursor() as cur:
            # 网络IO操作：执行SQL
            await cur.execute("SELECT Host,User FROM user")
            # 网络IO操作：获取SQL结果
            result = await cur.fetchall()
            print(result)
            # 网络IO操作：关闭链接
    return {"result": "ok"}
?
?
if __name__ == '__main__':
    uvicorn.run("fast2:app", host="127.0.0.1", port=5000, log_level="info")

(4) 爬虫

在编写爬虫应用时，需要通过网络IO去请求目标数据，这种情况适合使用异步编程来提升性能，接下来我们使用支持异步编程的aiohttp模块来实现。

安装aiohttp模块

pip3 install aiohttp

4-1 案例：基本使用

import aiohttp
import asyncio
?
async def fetch(session, url):
    print("发送请求：", url)
    async with session.get(url, verify_ssl=False) as response:
        text = await response.text()
        print("得到结果：", url, len(text))
?
async def main():
    async with aiohttp.ClientSession() as session:
        url_list = [
            'https://python.org',
            'https://www.baidu.com',
            'https://www.pythonav.com'
        ]
        loop = asyncio.get_event_loop()
        tasks = [loop.create_task(fetch(session, url)) for url in url_list]
        await asyncio.wait(tasks)
?
if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())