Python gevent协程模块详解：从原理到并发实战

陆景行

发布时间：2026-10-13 | 3人浏览过

原创

本文深入解析Python gevent模块，阐述协程与进程、线程的区别及上下文切换机制。重点讲解猴子补丁的使用、协程状态监控、超时控制、Event与Queue通信机制，以及基于BoundedSemaphore的信号量控制和协程本地变量。最后通过多并发Socket服务器与客户端示例，展示gevent在高并发网络编程中的实际应用。

Gevent官方文档地址：http://www.gevent.org/contents.html

进程、线程与协程的区别

通常所说的协程（Coroutine）源自corporate routine的缩写，直译为协同例程，一般简称为协程。

在Linux系统中，线程被视为轻量级进程，而协程则常被称为轻量级线程或微线程。

进程与协程的对比

以下对比进程和协程的异同点：

相同点：

当挂起一个执行流时，需要保存的内容包括：
栈空间：切换前需保存局部变量及函数调用链，否则无法恢复执行。

寄存器状态：用于记录执行流恢复后需执行的操作。

寄存器与栈的结合构成了上下文。上下文切换的理解如下： CPU看似并发执行多个进程，实则是通过处理器在进程间切换实现的。操作系统维持这种交错执行的机制称为上下文切换。

操作系统跟踪进程运行所需的所有状态信息，即上下文。任一时刻，操作系统仅能执行一个进程代码。当决定将控制权从当前进程转移至新进程时，需进行上下文切换：保存当前进程上下文，恢复新进程上下文，并将控制权移交新进程，使其从上次停止处继续执行。

不同点：

执行流调度者不同：进程由内核调度，协程在用户态调度。进程上下文在内核态保存恢复，协程在用户态保存恢复，显然用户态代价更低。
抢占机制不同：进程会被强占，而协程不会。若协程不主动让出CPU，其他协程将无法执行。
内存占用不同：协程仅需约4K栈空间即可运行，而进程占用内存大得多。
从操作系统视角看，多协程程序表现为单进程、单协程。

线程与协程的对比

前文已述协程亦称微线程，以下对比协程与线程：

线程间上下文切换成本相对较高，尤其在开启大量线程时；而协程切换成本极低。
线程切换多由操作系统控制，而协程执行由开发者自行控制。

协程仅在单一线程内不同协程间切换，与线程在进程下不同线程间切换类似，这也是协程被称为微线程的原因。

Gevent模块概述

Gevent是基于协程的Python网络库，利用Greenlet提供的封装了libevent事件循环的高层同步API。它使开发者无需改变编程习惯，即可用同步方式编写异步I/O代码。

简单示例：

import gevent
def test1():
  print 12
  gevent.sleep(0)
  print 34
def test2():
  print 56
  gevent.sleep(0)
  print 78
gevent.joinall([
  gevent.spawn(test1),
  gevent.spawn(test2),
])

结果：

12 56 34 78

猴子补丁（Monkey patching）

此补丁是Gevent模块的关键，使其发挥作用。我们通常使用Gevent实现网络通信的高并发，但Gevent直接修改标准库中大部分阻塞式系统调用（包括socket、ssl、threading和select等模块），使其变为协作式运行。然而，在复杂生产环境中，无法保证所有使用标准库的地方因打补丁而不出现异常问题。

一种方法是使用gevent下的socket模块，通过“from gevent import socket”导入。但更常用的方法是使用猴子补丁（Monkey patching）。尽管该方法褒贬不一，官网仍建议使用“patch_all()”，并在程序第一行执行。

from gevent import monkey; monkey.patch_socket()
import gevent
import socket
urls = ['www.baidu.com', 'www.gevent.org', 'www.python.org']
jobs = [gevent.spawn(socket.gethostbyname, url) for url in urls]
gevent.joinall(jobs, timeout=5)
print [job.value for job in jobs]

上述代码第一行对socket标准库打上猴子补丁，此后socket标准库中的类和方法均被替换为非阻塞式，其他代码无需修改，从而真正体现协程效率。Python中其他标准库也存在阻塞情况，gevent提供“monkey.patch_all()”方法替换所有标准库。

获取协程状态

started属性/ready()方法：判断协程是否已启动。
successful()方法：判断协程是否成功运行且未抛出异常。
value属性：获取协程执行完毕后的返回值。

此外，greenlet协程运行中发生的异常不会抛出到协程外，因此需通过协程对象的“exception”属性获取协程内异常。

以下示例演示了各种方法和属性的使用。

#！/usr/bin/env python
# _*_ coding utf-8 _*_
#Author: aaron
import gevent
def win():
  return 'You win!'
def fail():
 raise Exception('You failed!')
winner = gevent.spawn(win)
loser = gevent.spawn(fail)
print(winner.started) # True
print(loser.started)  # True
# 在Greenlet中发生的异常，不会被抛到Greenlet外面。
# 控制台会打出Stacktrace，但程序不会停止
try:
  gevent.joinall([winner, loser])
except Exception as e:
  # 这段永远不会被执行
  print('This will never be reached')
print(winner.ready()) # True
print(loser.started)  # True
print(winner.value) # 'You win!'
print(loser.value)  # None
print('successful ',winner.successful()) # True
print('successful ',loser.successful())  # False
# 这里可以通过raise loser.exception 或 loser.get()
# 来将协程中的异常抛出
print(loser.exception)

协程运行超时控制

此前介绍过在“gevent.joinall()”方法中可传入timeout参数设置超时，也可在全局范围内设置超时时间：

import gevent
from gevent import Timeout
timeout = Timeout(2) # 2 seconds
timeout.start()
def wait():

  gevent.sleep(10) 
try:
  gevent.spawn(wait).join()
except Timeout:
  print('Could not complete')

上例中，将超时设为2秒，此后所有协程运行若超过两秒将抛出“Timeout”异常。也可将超时设置在with语句内，使该设置仅在with语句块中有效：

with Timeout(1):
  gevent.sleep(10)

此外，可指定超时所抛出的异常以替换默认的“Timeout”异常。如下例中超时将抛出自定义的“TooLong”异常。

class TooLong(Exception):
  pass
with Timeout(1, TooLong):
  gevent.sleep(10)

协程间通信

事件（Event）对象

greenlet协程间异步通讯可使用事件（Event）对象。该对象的“wait()”方法可阻塞当前协程，而“set()”方法可唤醒之前阻塞的协程。在以下示例中，5个waiter协程均等待事件evt，当setter协程在3秒后设置evt事件，所有waiter协程即被唤醒。

#！/usr/bin/env python
# _*_ coding utf-8 _*_
#Author: aaron
import gevent
from gevent.event import Event
evt = Event()

def setter():
  print 'Wait for me'
  gevent.sleep(3) # 3秒后唤醒所有在evt上等待的协程
  print "Ok, I'm done"
  evt.set() # 唤醒

def waiter():
  print "I'll wait for you"
  evt.wait() # 等待
  print 'Finish waiting'

gevent.joinall([
  gevent.spawn(setter),
  gevent.spawn(waiter),
  gevent.spawn(waiter),
  gevent.spawn(waiter),
  gevent.spawn(waiter),
  gevent.spawn(waiter)
])

AsyncResult事件

除Event事件外，gevent还提供AsyncResult事件，可在唤醒时传递消息。将上例中的setter和waiter作如下改动：

#！/usr/bin/env python
# _*_ coding utf-8 _*_
#Author: aaron
from gevent.event import AsyncResult
aevt = AsyncResult()

def setter():
  print 'Wait for me'
  gevent.sleep(3) # 3秒后唤醒所有在evt上等待的协程
  print "Ok, I'm done"
  aevt.set('Hello!') # 唤醒，并传递消息

def waiter():
  print("I'll wait for you")
  message = aevt.get() # 等待，并在唤醒时获取消息
  print 'Got wake up message: %s' % message

队列（Queue）

队列Queue的概念众所周知，可通过put和get方法存取队列元素。gevent的队列对象允许greenlet协程间安全访问。运行以下程序，可见3个消费者分别消费队列中的产品，且消费过的产品不会被另一消费者再次取到：

#！/usr/bin/env python
# _*_ coding utf-8 _*_
#Author: aaron<br>
import gevent
from gevent.queue import Queue
products = Queue()
def consumer(name):
  #while not products.empty():
  while True:
    try:
      print('%s got product %s' % (name, products.get_nowait()))
      gevent.sleep(0)
    except gevent.queue.Empty:
      break
  print('Quit')

def producer():
  for i in range(1, 10):
    products.put(i)

gevent.joinall([
  gevent.spawn(producer),
  gevent.spawn(consumer, 'steve'),
  gevent.spawn(consumer, 'john'),
  gevent.spawn(consumer, 'nancy'),
])

注意：协程队列与线程队列相同，put和get方法均为阻塞式，它们均有非阻塞版本：put_nowait和get_nowait。若调用get方法时队列为空，不会抛出“gevent.queue.Empty”异常。只能使用get_nowait()方式使其抛出异常。

信号量

信号量可用于限制协程并发个数。它有两个方法：acquire和release。顾名思义，acquire即获取信号量，release即释放。当所有信号量均已被获取，剩余协程只能等待任一协程释放信号量后得以运行：

#！/usr/bin/env python
# _*_ coding utf-8 _*_
#Author: aaron
import gevent
from gevent.coros import BoundedSemaphore
sem = BoundedSemaphore(2)
def worker(n):
  sem.acquire()
  print('Worker %i acquired semaphore' % n)
  gevent.sleep(0)
  sem.release()
  print('Worker %i released semaphore' % n)
gevent.joinall([gevent.spawn(worker, i) for i in xrange(0, 6)])

上例中，初始化了“BoundedSemaphore”信号量，并将其个数定为2。因此同一时间，仅有两个worker协程被调度。程序运行后结果如下：

Worker 0 acquired semaphore Worker 1 acquired semaphore Worker 0 released semaphore Worker 1 released semaphore Worker 2 acquired semaphore Worker 3 acquired semaphore Worker 2 released semaphore Worker 3 released semaphore Worker 4 acquired semaphore Worker 4 released semaphore Worker 5 acquired semaphore Worker 5 released semaphore

若信号量个数为1，则等同于同步锁。

协程本地变量

与线程类似，协程也有本地变量，即仅在当前协程内可访问的变量：

#！/usr/bin/env python
# _*_ coding utf-8 _*_
#Author: aaron
import gevent
from gevent.local import local
data = local()
def f1():
  data.x = 1
  print data.x

def f2():
  try:
    print data.x
  except AttributeError:
    print 'x is not visible'

gevent.joinall([
  gevent.spawn(f1),
  gevent.spawn(f2)
])

通过将变量存放在local对象中，即可将其作用域限制在当前协程内。当其他协程尝试访问该变量时，将抛出异常。不同协程间可有重名的本地变量，且互不影响。因协程本地变量的实现，是将其存放在以“greenlet.getcurrent()”的返回为键值的私有命名空间内。

多并发Socket模型

服务器端：

#！/usr/bin/env python
# _*_ coding utf-8 _*_
#Author: aaron
import socket
import gevent
from gevent import socket, monkey
monkey.patch_all()
def server(port):
  s = socket.socket()
  s.bind(('0.0.0.0', port))
  s.listen(500)
  while True:
    cli, addr = s.accept()
    gevent.spawn(handle_request, cli)

def handle_request(conn):
  try:
    while True:
      data = conn.recv(1024)
      print("recv:", data)
      conn.send(data)
      if not data:
        conn.shutdown(socket.SHUT_WR)

  except Exception as ex:
    print(ex)
  finally:
    conn.close()
if __name__ == '__main__':
  server(8001)

当客户端连接上服务器端时，服务器端通过开辟一个协程与该客户端完成交互任务。同时，由于使用了Gevent协程方式，在每个客户端与服务器交互时，并不会影响服务器端的工作。

客户端：

#！/usr/bin/env python
# _*_ coding utf-8 _*_
#Author: aaron
import socket
HOST = 'localhost' # The remote host
PORT = 8001     # The same port as used by the server
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.connect((HOST, PORT))
while True:
  msg = bytes(input(">>:"), encoding="utf8")
  s.sendall(msg)
  data = s.recv(1024)
  # print(data)
  print('Received', repr(data)) # repr 格式化输出
s.close()

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持本站。