单线程并发实现：探索CPU效率提升与任务切换保存状态的新策略

前引

之前我们学习了线程和进程的概念，知道在操作系统中，进程是资源分配的最小单位，而线程是 CPU 调度的最小单位。按理而言，我们已经将 CPU 的利用率提高了不少。但是我们清楚，无论是创建多进程还是创建多线程来处理问题，都需要花费一定的时间去创建进程、创建线程以及管理它们之间的切换。

随着我们越发追求效率，基于单线程来实现并发又成为一个新的问题，也就是在只有一个主线程（很明显能利用的 CPU 只有一个）的情况下实现并发。这样就能节省创建线程进程所耗费的时间。

为此我们需要先回顾下并发的本质：切换+保存状态

cpu 正在运行一个任务，在以下两种情况下会切走去执行其他任务，且切换由操作系统强制控制。其一，该任务发生了阻塞；其二，该任务计算的时间过长。

在介绍进程理论时，提到了进程的三种执行状态。因为线程才是执行单位，所以也可以把上图理解为线程的三种状态。

其中第二种情况无法提升效率，只是为了让 cpu 能够公平对待各个任务，达到看似所有任务都在“同时”执行的效果。倘若多个任务都是纯粹的计算任务，那么这种任务切换反而会致使效率降低。

为此我们能够依据 yield 进行验证。yield 本身是一种在单线程环境下能够留存任务运行状态的方法，我们来简略地复习一下。

yield 能够保存状态，它的状态保存与操作系统保存线程状态相似，不过 yield 是由代码级别来进行控制的，并且更加轻量级。
send 能够将一个函数的结果传递给另一个函数，通过这种方式可以在单线程内实现程序之间的切换。

单纯地切换反而会降低运行效率

#串行执行
import time
def consumer(res):
    '''任务1:接收数据,处理数据'''
    pass
def producer():
    '''任务2:生产数据'''
    res=[]
    for i in range(10000000):
        res.append(i)
    return res
start=time.time()
#串行执行
res=producer()
将 consumer(res)写成 consumer(producer())会致使执行效率降低。
stop=time.time()
打印(停止 - 开始)  # 1.5536692142486572
#基于yield并发执行
import time
def consumer():
    '''任务1:接收数据,处理数据'''
    while True:
        x=yield
def producer():
    '''任务2:生产数据'''
    g=consumer()
    next(g)
    for i in range(10000000):
        g.send(i)
start=time.time()
基于 yield 来保存状态，这样就能实现两个任务之间直接进行来回切换，也就达到了并发的效果。
如果每个任务中都加入打印操作，那么可以明显地看到两个任务的打印呈现出你一次我一次的情况，这也就意味着两个任务是并发执行的。
producer()
stop=time.time()
打印(停止 - 开始)

第一种情况的切换：当任务一遇到 io 时，会切换到任务二去执行。通过这种方式，可以利用任务一阻塞的时间来完成任务二的计算，效率提升的关键就在于此。

yield无法做到遇到io阻塞

import time
def consumer():
    '''任务1:接收数据,处理数据'''
    while True:
        x=yield
def producer():
    '''任务2:生产数据'''
    g=consumer()
    next(g)
    for i in range(10000000):
        g.send(i)
        time.sleep(2)
start=time.time()
生产者()#会并发执行，然而当生产者任务遇到输入输出操作时就会被阻塞，不会切换到该线程内的其他任务去执行。
stop=time.time()
print(stop-start)

对于单线程序，程序中不可避免会出现 io 操作。如果能在自己的程序（也就是用户程序级别，而非操作系统级别）控制单线程下的多个任务，当一个任务遇到 io 阻塞时就切换到另一个任务去计算，那么就可以保证该线程最大限度地处于就绪态，也就是随时可以被 cpu 执行的状态。这相当于在用户程序级别将自己的 io 操作最大限度地隐藏起来，进而能够迷惑操作系统，让操作系统看到：该线程好像一直在计算，io 比较少，从而能让操作系统更多地将 cpu 的执行权限分配给我们的线程。

协程的本质在于单线程环境中，由用户自行掌控。当一个任务遭遇 io 阻塞时，便可切换至另一个任务去执行，通过这种方式来提高效率。为达成此目的，我们需要找到一种能够同时满足下述条件的解决方案。

能够控制多个任务的切换，在切换之前会把任务的状态保存起来，这样在重新运行时，就可以依据暂停的位置继续进行执行。
作为对 1 的补充内容：能够检测输入输出（io）操作，并且只有在遇到 io 操作的情形下才会发生切换。

协程介绍

协程是单线程下的并发形式，也被称作微线程或纤程。它的英文名是 Coroutine。可以这样说协程：协程属于用户态的轻量级线程，也就是说协程是由用户程序自身来进行控制调度的。

需要强调的是：

python 的线程属于内核级别。这意味着它由操作系统来控制调度。例如，当单线程遇到 io 操作或者执行时间过长时，就会被迫交出 cpu 的执行权限，然后切换其他线程运行。
单线程内开启协程，当遇到 io 时，会从应用程序级别进行控制切换，而不是从操作系统级别。通过这种方式来提升效率，需要注意的是，非 io 操作的切换与效率无关。

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

优点

协程的切换开销比较小，它属于程序级别的切换，操作系统完全无法感知，所以它更加轻量级。
#2. 单线程内就可以实现并发的效果，最大限度地利用cpu

缺点

协程的本质在于单线程环境下无法利用多核。可以通过一个程序开启多个进程，在每个进程内开启多个线程，接着在每个线程内开启协程。
协程指的是单个线程。所以一旦协程出现阻塞的情况，就会阻塞整个线程。

协程特点

在只有一个单线程中实现并发修改共享数据时，无需加锁，用户程序需自己保存多个控制流的上下文栈。另外，一个协程遇到 IO 操作会自动切换到其他协程，这里检测 IO 的实现方式，yield 和 greenlet 都无法做到，所以用到了 gevent 模块（select 机制）。

安装

pip3 install greenlet

greenlet实现状态切换

from greenlet import greenlet
def eat(name):
    print('%s eat 1' %name)
    g2.switch('egon')
    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)
可以在第一次进行 switch 操作时传入参数，之后就不需要再传入参数了。

单纯的进行切换（在不存在 io 的情形下或者没有再次开辟内存空间的操作时），这样反而会让程序的执行速度降低。

效率对比

#顺序执行
import time
def f1():
    res=1
    for i in range(100000000):
        res+=i
def f2():
    res=1
    for i in range(100000000):
        res*=i
start=time.time()
f1()
f2()
stop=time.time()
打印('运行时间是 %s' % (停止 - 开始))
#切换
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('运行时间是 %s' % (停止时间 - 开始时间))  # 52.763017892837524

greenlet 仅提供了一种相较于 generator 更为便捷的切换方式。当切换到一个任务执行时，倘若遇到 io，就会在原地阻塞。即便如此，它依然没有解决遇到 IO 自动切换以提升效率的问题。

单线程里的这 20 个任务的代码通常包含计算操作和阻塞操作。在执行任务 1 时若遇到阻塞，我们能够利用阻塞的时间去执行任务 2。如此这般，便可提高效率，而这就用到了 Gevent 模块。

Gevent模块

安装

pip3 install gevent

Gevent 是一个第三方库。它能够轻松借助 gevent 来实现并发同步或异步编程。在 gevent 中，主要用到的模式是 Greenlet。Greenlet 是以 C 扩展模块的形式接入 Python 的轻量级协程。Greenlet 全部在主程序操作系统进程的内部运行，并且它们是被协作式地进行调度的。

用法

创建一个协程对象 g1，g1 由 gevent.spawn 函数创建。spawn 括号内的第一个参数是函数名，例如 eat。后面可以有多个参数，这些参数既可以是位置实参，也可以是关键字实参，并且都会传递给函数 eat。
g2=gevent.spawn(func2)
g1.join() #等待g1结束
g2.join() #等待g2结束
可以将上述两步合并为一步，即 gevent.joinall([g1, g2])
g1.value#拿到func1的返回值

例：遇到io主动切换

import gevent
def eat(name):
    print('%s eat 1' %name)
    gevent.sleep(2)
    print('%s eat 2' %name)
def play(name):
    print('%s play 1' %name)
    gevent.sleep(1)
    print('%s play 2' %name)
g1=gevent.spawn(eat,'egon')
g2 是通过 gevent 启动一个名为 play 且名字为 'egon' 的进程
g1.join()
g2.join()
#或者gevent.joinall([g1,g2])
print('主')

上例中 gevent.sleep(2)所模拟的是 gevent 能够识别的 io 阻塞。然而，对于 time.sleep(2)或者其他类型的阻塞，gevent 是无法直接识别的。要让 gevent 能够识别这些阻塞，就需要使用下面这一行代码来打补丁。

from gevent import monkey; 必须将 monkey.patch_all()放在被打补丁的模块之前，例如 time 模块之前，以及 socket 模块之前。

我们可以这样记忆：若要使用 gevent，就需要把 from gevent import monkey;monkey.patch_all()放置在文件的开头。

导入 gevent 库中的 monkey 模块，然后调用 monkey.patch_all()函数进行相关的补丁操作。
import gevent
import time
def eat():
    print('eat food 1')
    time.sleep(2)
    print('eat food 2')
def play():
    print('play 1')
    time.sleep(1)
    print('play 2')
g1=gevent.spawn(eat)
g2=gevent.spawn(play)
gevent.joinall([g1,g2])
print('主')

我们能够使用 threading.current_thread().getName()去查看每一个 g1 以及每一个 g2 ，查看所得到的结果是 DummyThread-n ，也就是假线程

查看当前线程的名称，即 threading.current_thread().getName() 所获取的内容。

from gevent import monkey;monkey.patch_all()
import threading
import gevent
import time
def eat():
输出当前线程的名称。通过 threading.current_thread().getName() 来获取当前线程的名称并进行输出。
    print('eat food 1')
    time.sleep(2)
    print('eat food 2')
def play():
    print(threading.current_thread().getName())
    print('play 1')
    time.sleep(1)
    print('play 2')
g1=gevent.spawn(eat)
g2=gevent.spawn(play)
gevent.joinall([g1,g2])
print('主')

Gevent之同步与异步

使用 gevent 库中的 spawn 函数、joinall 函数和 monkey 模块；并且调用 monkey.patch_all()方法来进行相关的补丁操作。
import time
def task(pid):
    """
    Some non-deterministic task
    """
    time.sleep(0.5)
    print('Task %s done' % pid)
def synchronous():  # 同步
    for i in range(10):
        task(i)
def asynchronous(): # 异步
g_l 等于通过对 range(10) 中的每个 i 执行 spawn(task,i) 操作所生成的列表。
    joinall(g_l)
    print('DONE')
    
if __name__ == '__main__':
    print('Synchronous:')
    synchronous()
    print('Asynchronous:')
    asynchronous()
上面程序的一个重要部分是把 task 函数封装到 Greenlet 内部线程当中，使用了 gevent.spawn 这个方法。
初始化的 greenlet 列表被存放在数组 threads 里，这个数组被传递给了 gevent.joinall 函数。
后者会阻塞当前的流程，并且会执行所有被给定的 greenlet 任务。只有在所有的 greenlet 任务都执行完毕之后，执行流程才会继续向下进行。

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)
如果 response 的状态码等于 200 。
输出从指定的 url 接收到的字节数，这个字节数是 response.text 的长度，具体格式为 '%d bytes received from %s'，其中 %d 被替换为 len(response.text) 的值，%s 被替换为 url 的值
start_time=time.time()
gevent.joinall([
gevent 启动一个新的协程来执行 get_page 函数，并传入参数 'https://www.python.org/' 。
gevent.spawn 函数用于启动一个新的协程来获取页面，获取的页面地址为 'https://www.yahoo.com/'。
gevent 启动一个新的协程来获取页面，获取的页面地址是 'https://github.com/' 。
])
stop_time=time.time()
打印('运行时间是 %s' % (停止时间-开始时间))

Gevent之应用举例二

通过gevent实现单线程下的socket并发

要将 from gevent import monkey;monkey.patch_all()放到导入 socket 模块之前，不然 gevent 就无法识别 socket 的阻塞。

server

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 对 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)
打印('客户端 %s:%s 消息: %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)

client

from socket import *
创建一个客户端，通过调用 socket 函数来实现。该函数的第一个参数 AF_INET 表示使用 IPv4 地址族，第二个参数 SOCK_STREAM 表示使用流式套接字。即客户端通过 socket(AF_INET,SOCK_STREAM)来建立与服务器的连接。
客户端连接到了（'127.0.0.1'，8080）这个地址
while True:
    msg=input('>>: ').strip()
    if not msg:continue
客户端发送以 utf-8 编码的 msg 。
    msg=client.recv(1024)
    print(msg.decode('utf-8'))

多线程并发多个客户端

from threading import Thread
from socket import *
import threading
def client(server_ip,port):
c 等于创建一个套接字对象，该套接字使用的地址族是 AF_INET，套接字类型是 SOCK_STREAM。套接字对象一定要被添加到函数内部，也就是局部名称空间内。如果将其放在函数外部，那么这个套接字对象就会被所有线程共享，这样一来，客户端的端口就会始终保持一样。
    c.connect((server_ip,port))
    count=0
    while True:
c 发送将字符串（由 threading.current_thread().getName() 和 count 组成的 '%s say hello %s' 格式的字符串）编码为 utf-8 后的内容。即 c 发送（（当前线程的名称 say hello 计数）编码后的内容）。这里的计数为 count ，当前线程的名称通过 threading.current_thread().getName() 获取。
        msg=c.recv(1024)
        print(msg.decode('utf-8'))
        count+=1
if __name__ == '__main__':
    for i in range(500):
创建一个线程 t，其目标函数是 client，参数为('127.0.0.1', 8080)
        t.start()