Gevent 中文入门教程

核心

Greenlets

gevent中使用的主要模式是 Greenlet，这是作为C扩展模块提供给 Python 的一个轻量级协同程序。所有 greenlet 都在主程序的 OS 进程中运行，但它们是协同调度的。

在任何给定的时间内，只有一个greenlet在运行。

这不同于由 multiprocessing 或 threading 库提供的并行结构，它们这些库可以自旋进程和POSIX线程，由操作系统调度，并且真正并行的。

异步和同步执行

并发性的核心思想是，可以将较大的任务分解为多个子任务的集合，这些子任务计划同时或异步运行，而不是一次或同步运行。两个子任务之间的切换称为上下文切换。

gevent中的上下文切换是通过 yielding 来完成的。在本例中，我们有两个上下文，它们通过调用 gevent.sleep(0) 相互让步。

  1. import gevent
  3. def foo():
  4.     print('Running in foo')
  5.     gevent.sleep(0)
  6.     print('Explicit context switch to foo again')
  8. def bar():
  9.     print('Explicit context to bar')
  10.     gevent.sleep(0)
  11.     print('Implicit context switch back to bar')
  13. gevent.joinall([
  14.     gevent.spawn(foo),
  15.     gevent.spawn(bar),
  16. ])
  1. Running in foo
  2. Explicit context to bar
  3. Explicit context switch to foo again
  4. Implicit context switch back to bar

当我们将 gevent 用于网络和 IO 绑定函数时，它的真正威力就来了，这些函数可以协同调度。Gevent 负责处理所有细节，以确保网络库尽可能隐式地让步出它们的 greenlet 上下文。我再怎么强调这是一个多么有力的成语也不为过。但也许可以举个例子来说明。

在这种情况下，select() 函数通常是一个阻塞调用，它轮询各种文件描述符。

  1. import time
  2. import gevent
  3. from gevent import select
  5. start = time.time()
  6. tic = lambda: 'at %1.1f seconds' % (time.time() - start)
  8. def gr1():
  9.     # Busy waits for a second, but we don't want to stick around...
  10.     print('Started Polling: %s' % tic())
  11.     select.select([], [], [], 2)                 
  12.     # 可以理解成一个 IO 阻塞的操作，阻塞了2秒，这时 Greenlet 会自动切换到 gr2() 上下文执行 
  13.     print('Ended Polling: %s' % tic())
  15. def gr2():
  16.     # Busy waits for a second, but we don't want to stick around...
  17.     print('Started Polling: %s' % tic())
  18.     select.select([], [], [], 2)
  19.     print('Ended Polling: %s' % tic())
  21. def gr3():
  22.     print("Hey lets do some stuff while the greenlets poll, %s" % tic())
  23.     gevent.sleep(1)
  24.     # 让当前 Greenlet 休眠1秒，上面 gr1() gr2() 阻塞操作完成后，再切换到 gr1() gr2() 的上下文执行
  26. gevent.joinall([
  27.     gevent.spawn(gr1),
  28.     gevent.spawn(gr2),
  29.     gevent.spawn(gr3),
  30. ])
  1. Started Polling: at 0.0 seconds
  2. Started Polling: at 0.0 seconds
  3. Hey lets do some stuff while the greenlets poll, at 0.0 seconds
  4. Ended Polling: at 2.0 seconds
  5. Ended Polling: at 2.0 seconds

另一个比较综合的例子定义了一个不确定的任务函数 (它的输出不能保证对相同的输入给出相同的结果) 。在这种情况下，运行该函数的副作用是任务暂停执行的时间是随机的。

  1. import gevent
  2. import random
  4. def task(pid):
  5.     """
  6.     Some non-deterministic task
  7.     """
  8.     gevent.sleep(random.randint(0,2)*0.001)
  9.     print('Task %s done' % pid)
  11. def synchronous():
  12.     for i in range(1,10):
  13.         task(i)
  15. def asynchronous():
  16.     threads = [gevent.spawn(task, i) for i in xrange(10)]
  17.     gevent.joinall(threads)
  19. print('Synchronous:')
  20. synchronous()
  22. print('Asynchronous:')
  23. asynchronous()
  1. Synchronous:
  2. Task 1 done
  3. Task 2 done
  4. Task 3 done
  5. Task 4 done
  6. Task 5 done
  7. Task 6 done
  8. Task 7 done
  9. Task 8 done
  10. Task 9 done
  11. Asynchronous:
  12. Task 1 done
  13. Task 5 done
  14. Task 6 done
  15. Task 2 done
  16. Task 4 done
  17. Task 7 done
  18. Task 8 done
  19. Task 9 done
  20. Task 0 done
  21. Task 3 done

在同步情况下，所有任务都是按顺序运行的，这会导致在执行每个任务时主程序阻塞(即暂停主程序的执行)。

程序的重要部分是 gevent.spawn，它将给定的函数封装在一个 Greenlet 线程中。初始化的 greenlet 列表存储在传递给 gevent 的数组线程中。gevent.joinall 函数，它会阻塞当前程序，来运行所有给定的 greenlet。只有当所有 greenlet 终止时，执行才会继续进行。

需要注意的是，异步情况下的执行顺序本质上是随机的，异步情况下的总执行时间比同步情况下少得多。实际上，同步的例子完成的最大时间是每个任务停顿0.002秒，导致整个队列停顿0.02秒。在异步情况下，最大运行时间大约为0.002秒，因为没有一个任务会阻塞其他任务的执行。

在更常见的用例中，异步地从服务器获取数据，fetch() 的运行时在请求之间会有所不同，这取决于请求时远程服务器上的负载。

  1. import gevent.monkey
  2. gevent.monkey.patch_socket()
  3. # 把内置的阻塞的 socket替换成非阻塞的socket
  5. import gevent
  6. import urllib2
  7. import simplejson as json
  9. def fetch(pid):
  10.     response = urllib2.urlopen('http://json-time.appspot.com/time.json')
  11.     result = response.read()
  12.     json_result = json.loads(result)
  13.     datetime = json_result['datetime']
  15.     print('Process %s: %s' % (pid, datetime))
  16.     return json_result['datetime']
  18. def synchronous():
  19.     for i in range(1,10):
  20.         fetch(i)
  22. def asynchronous():
  23.     threads = []
  24.     for i in range(1,10):
  25.         threads.append(gevent.spawn(fetch, i))
  26.     gevent.joinall(threads)
  28. print('Synchronous:')
  29. synchronous()
  31. print('Asynchronous:')
  32. asynchronous()

确定性

如前所述，greenlet 是确定的。给定相同的 greenlet 配置和相同的输入集，它们总是产生相同的输出。例如，让我们将一个任务分散到一个多进程（multiprocessing）池中，并将其结果与一个gevent池的结果进行比较。

  1. import time
  3. def echo(i):
  4.     time.sleep(0.001)
  5.     return i
  7. # Non Deterministic Process Pool
  9. from multiprocessing.pool import Pool
  11. p = Pool(10)
  12. run1 = [a for a in p.imap_unordered(echo, xrange(10))]          # imap_unordered 返回一个顺序随机的 iterable 对象
  13. run2 = [a for a in p.imap_unordered(echo, xrange(10))]
  14. run3 = [a for a in p.imap_unordered(echo, xrange(10))]
  15. run4 = [a for a in p.imap_unordered(echo, xrange(10))]
  17. print(run1 == run2 == run3 == run4)
  19. # Deterministic Gevent Pool
  21. from gevent.pool import Pool
  23. p = Pool(10)
  24. run1 = [a for a in p.imap_unordered(echo, xrange(10))]      # [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
  25. run2 = [a for a in p.imap_unordered(echo, xrange(10))]      # [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
  26. run3 = [a for a in p.imap_unordered(echo, xrange(10))]      # [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
  27. run4 = [a for a in p.imap_unordered(echo, xrange(10))]      # [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
  29. print(run1 == run2 == run3 == run4)
  1. False
  2. True

尽管 gevent 通常是确定的，但当您开始与外部服务(如 socket 和文件)进行交互时，非确定性的来源可能会潜入您的程序。因此，即使 green 线程是确定性并发的一种形式，它们仍然会遇到POSIX线程和进程遇到的一些相同的问题。

与并发有关的长期问题称为竞争条件。简而言之，当两个并发线程/进程依赖于某些共享资源但还试图修改该值时，就会发生竞争状态。这将导致资源的值变得依赖于执行顺序。这是一个问题，一般来说，应该尽量避免竞态条件，因为它们会导致全局的不确定程序行为。

最好的方法是在任何时候都避免所有全局状态。全局状态和导入时间的副作用总是会回来咬你一口

生成 Greenlets

gevent提供了一些关于Greenlet初始化的包装器。一些最常见的模式是：

  1. import gevent
  2. from gevent import Greenlet
  4. def foo(message, n):
  5.     """
  6.     Each thread will be passed the message, and n arguments
  7.     in its initialization.
  8.     """
  9.     gevent.sleep(n)
  10.     print(message)
  12. # Initialize a new Greenlet instance running the named function
  13. # foo
  14. thread1 = Greenlet.spawn(foo, "Hello", 1)
  16. # Wrapper for creating and running a new Greenlet from the named
  17. # function foo, with the passed arguments
  18. thread2 = gevent.spawn(foo, "I live!", 2)
  20. # Lambda expressions
  21. thread3 = gevent.spawn(lambda x: (x+1), 2)
  23. threads = [thread1, thread2, thread3]
  25. # Block until all threads complete.
  26. gevent.joinall(threads)
  1. Hello
  2. I live!

除了使用基本的Greenlet类，您还可以子类化 Greenlet 类并覆盖 _run 方法。

  1. import gevent
  2. from gevent import Greenlet
  4. class MyGreenlet(Greenlet):
  6.     def __init__(self, message, n):
  7.         Greenlet.__init__(self)
  8.         self.message = message
  9.         self.n = n
  11.     def _run(self):
  12.         print(self.message)
  13.         gevent.sleep(self.n)
  15. g = MyGreenlet("Hi there!", 3)
  16. g.start()
  17. g.join()
  1. Hi there!

Greenlets 状态

与代码的其他部分一样，greenlet可能以各种方式失败。greenlet可能无法抛出异常、无法停止或消耗太多系统资源。

greenlet 的内部状态通常是一个与时间相关的参数。在greenlets上有许多标志，它们允许您监视线程的状态：

started — 布尔值，指示Greenlet是否已启动

ready() — 布尔值，指示Greenlet是否已停止

successful() — 布尔值，指示Greenlet是否已停止且没有抛出异常

value — Greenlet返回的值

exception — 异常，在greenlet中抛出的未捕获异常实例

  1. import gevent
  3. def win():
  4.     return 'You win!'
  6. def fail():
  7.     raise Exception('You fail at failing.')
  9. winner = gevent.spawn(win)
  10. loser = gevent.spawn(fail)
  12. print(winner.started) # True
  13. print(loser.started)  # True
  15. # Exceptions raised in the Greenlet, stay inside the Greenlet.
  16. try:
  17.     gevent.joinall([winner, loser])
  18. except Exception as e:
  19.     print('This will never be reached')
  21. print(winner.value) # 'You win!'
  22. print(loser.value)  # None
  24. print(winner.ready()) # True
  25. print(loser.ready())  # True
  27. print(winner.successful()) # True
  28. print(loser.successful())  # False
  30. # The exception raised in fail, will not propagate outside the
  31. # greenlet. A stack trace will be printed to stdout but it
  32. # will not unwind the stack of the parent.
  34. print(loser.exception)
  36. # It is possible though to raise the exception again outside
  37. # raise loser.exception
  38. # or with
  39. # loser.get()
  1. True
  2. True
  3. You win!
  4. None
  5. True
  6. True
  7. True
  8. False
  9. You fail at failing.

程序关闭

当主程序接收到SIGQUIT时，无法生成（yield）的 greenlet 可能会使程序的执行时间比预期的更长。这将导致所谓的“僵尸进程”，需要从 Python 解释器外部杀死这些进程。

一种常见的模式是监听主程序上的SIGQUIT事件并在退出前调用 gevent.shutdown 。

  1. import gevent
  2. import signal
  4. def run_forever():
  5.     gevent.sleep(1000)
  7. if __name__ == '__main__':
  8.     gevent.signal(signal.SIGQUIT, gevent.kill)
  9.     thread = gevent.spawn(run_forever)
  10.     thread.join()

超时

超时是对代码块或Greenlet的运行时的约束。

  1. import gevent
  2. from gevent import Timeout
  4. seconds = 10
  6. timeout = Timeout(seconds)
  7. timeout.start()
  9. def wait():
  10.     gevent.sleep(10)
  12. try:
  13.     gevent.spawn(wait).join()
  14. except Timeout:
  15.     print('Could not complete')

在with语句中，它们还可以与上下文管理器一起使用。

  1. import gevent
  2. from gevent import Timeout
  4. time_to_wait = 5 # seconds
  6. class TooLong(Exception):
  7.     pass
  9. with Timeout(time_to_wait, TooLong):
  10.     gevent.sleep(10)

此外，gevent 还为各种 Greenlet 和数据结构相关的调用提供超时参数。例如：

  1. import gevent
  2. from gevent import Timeout
  4. def wait():
  5.     gevent.sleep(2)
  7. timer = Timeout(1).start()
  8. thread1 = gevent.spawn(wait)
  10. try:
  11.     thread1.join(timeout=timer)
  12. except Timeout:
  13.     print('Thread 1 timed out')
  15. # --
  17. timer = Timeout.start_new(1)
  18. thread2 = gevent.spawn(wait)
  20. try:
  21.     thread2.get(timeout=timer)
  22. except Timeout:
  23.     print('Thread 2 timed out')
  25. # --
  27. try:
  28.     gevent.with_timeout(1, wait)
  29. except Timeout:
  30.     print('Thread 3 timed out')
  1. Thread 1 timed out
  2. Thread 2 timed out
  3. Thread 3 timed out

猴子补丁

我们来到了Gevent的黑暗角落。到目前为止，我一直避免提到monkey patching，以尝试和激发强大的协同模式，但是现在是讨论monkey patching的黑魔法的时候了。 如果您注意到上面我们调用了命令 monkey.patch_socket()，这是一个纯粹用于修改标准库套接字库(socket)的副作用命令。

  1. import socket
  2. print(socket.socket)
  4. print("After monkey patch")
  5. from gevent import monkey
  6. monkey.patch_socket()
  7. print(socket.socket)
  9. import select
  10. print(select.select)
  11. monkey.patch_select()
  12. print("After monkey patch")
  13. print(select.select)
  1. class 'socket.socket'
  2. After monkey patch
  3. class 'gevent.socket.socket'
  5. built-in function select
  6. After monkey patch
  7. function select at 0x1924de8

Python 允许在运行时修改大多数对象，包括模块、类甚至函数。这通常是一个令人震惊的坏主意，因为它创建了一个“隐式副作用”，如果出现问题，通常非常难以调试，然而在极端情况下，库需要改变Python本身的基本行为，可以使用monkey补丁。在这种情况下，gevent能够修补标准库中的大多数阻塞系统调用，包括 socket、ssl、threading 和 select 模块中的调用。

例如，Redis-python 的绑定通常使用常规tcp socket与Redis-server实例通信。只需调用gevent.monkey.patch_all()，我们可以让redis绑定协同调度请求，并与gevent堆栈的其他部分一起工作。

这让我们可以在不编写任何代码的情况下集成通常无法与gevent一起工作的库。（尽管猴子补丁仍然是邪恶的，但在这种情况下，它是“有用的邪恶”。）

数据结构

事件

事件是greenlet之间异步通信的一种形式。

  1. import gevent
  2. from gevent.event import Event
  4. '''
  5. Illustrates the use of events
  6. '''
  9. evt = Event()
  11. def setter():
  12.     '''After 3 seconds, wake all threads waiting on the value of evt'''
  13.     print('A: Hey wait for me, I have to do something')
  14.     gevent.sleep(3)
  15.     print("Ok, I'm done")
  16.     evt.set()   # 运行到evt.set()会将flag设置为True，然后另外两个被阻塞的waitter的evt.wait()方法在看到flag已经为True之后不再继续阻塞运行并且结束。
  19. def waiter():
  20.     '''After 3 seconds the get call will unblock'''
  21.     print("I'll wait for you")
  22.     evt.wait()  # blocking
  23.     print("It's about time")
  25. def main():
  26.     gevent.joinall([
  27.         gevent.spawn(setter),
  28.         gevent.spawn(waiter),
  29.         gevent.spawn(waiter),
  30.         gevent.spawn(waiter),
  31.         gevent.spawn(waiter),
  32.         gevent.spawn(waiter)
  33.     ])
  35. if __name__ == '__main__': main()

事件对象的扩展是 AsyncResult，它允许您在唤醒调用时发送一个值。这有时被称为future或deferred，因为它有对 future 值的引用，可以在任意的时间设置该值。

  1. import gevent
  2. from gevent.event import AsyncResult
  3. a = AsyncResult()
  5. def setter():
  6.     """
  7.     After 3 seconds set the result of a.
  8.     """
  9.     gevent.sleep(3)
  10.     a.set('Hello!')
  12. def waiter():
  13.     """
  14.     After 3 seconds the get call will unblock after the setter
  15.     puts a value into the AsyncResult.
  16.     """
  17.     print(a.get())
  19. gevent.joinall([
  20.     gevent.spawn(setter),
  21.     gevent.spawn(waiter),
  22. ])

队列

队列是按顺序排列的数据集，它们具有通常的 put / get 操作，但可以在Greenlets上安全操作的方式编写。

例如，如果一个Greenlet从队列中获取一个项目(item)，则同一项目(item)不会被同时执行的另一个Greenlet获取。

  1. import gevent
  2. from gevent.queue import Queue
  4. tasks = Queue()
  6. def worker(n):
  7.     while not tasks.empty():
  8.         task = tasks.get()
  9.         print('Worker %s got task %s' % (n, task))
  10.         gevent.sleep(0)
  12.     print('Quitting time!')
  14. def boss():
  15.     for i in xrange(1,25):
  16.         tasks.put_nowait(i)
  18. gevent.spawn(boss).join()
  20. gevent.joinall([
  21.     gevent.spawn(worker, 'steve'),
  22.     gevent.spawn(worker, 'john'),
  23.     gevent.spawn(worker, 'nancy'),
  24. ])
  1. Worker steve got task 1
  2. Worker john got task 2
  3. Worker nancy got task 3
  4. Worker steve got task 4
  5. Worker john got task 5
  6. Worker nancy got task 6
  7. Worker steve got task 7
  8. Worker john got task 8
  9. Worker nancy got task 9
  10. Worker steve got task 10
  11. Worker john got task 11
  12. Worker nancy got task 12
  13. Worker steve got task 13
  14. Worker john got task 14
  15. Worker nancy got task 15
  16. Worker steve got task 16
  17. Worker john got task 17
  18. Worker nancy got task 18
  19. Worker steve got task 19
  20. Worker john got task 20
  21. Worker nancy got task 21
  22. Worker steve got task 22
  23. Worker john got task 23
  24. Worker nancy got task 24
  25. Quitting time!
  26. Quitting time!
  27. Quitting time!

根据需要，队列还可以在put或get上阻塞。

每个put和get操作都有一个非阻塞的对应操作，put_nowait 和 get_nowait 不会阻塞。如果操作是不可能的，会引发 gevent.queue.Empty 或 gevent.queue.Full

在这个例子中，我们让boss同时和workers运行，并且对队列有一个限制，防止它包含三个以上的元素。这个限制意味着put操作将阻塞，直到队列上有空间为止。相反，如果队列上没有要获取的元素，get操作就会阻塞，它还会接受一个超时参数，如果在超时的时间范围内找不到工作(work)，则允许队列以异常 gevent.queue.Empty 退出。

  1. import gevent
  2. from gevent.queue import Queue, Empty
  4. tasks = Queue(maxsize=3)
  6. def worker(name):
  7.     try:
  8.         while True:
  9.             task = tasks.get(timeout=1) # decrements queue size by 1
  10.             print('Worker %s got task %s' % (name, task))
  11.             gevent.sleep(0)
  12.     except Empty:
  13.         print('Quitting time!')
  15. def boss():
  16.     """
  17.     Boss will wait to hand out work until a individual worker is
  18.     free since the maxsize of the task queue is 3.
  19.     """
  21.     for i in xrange(1,10):
  22.         tasks.put(i)                            # 输入1，2，3，到4时，队列达到最大值，put方法阻塞，gevent 切换到下一个协程worker（steve）
  23.     print('Assigned all work in iteration 1')
  25.     for i in xrange(10,20):
  26.         tasks.put(i)
  27.     print('Assigned all work in iteration 2')
  29. gevent.joinall([
  30.     gevent.spawn(boss),
  31.     gevent.spawn(worker, 'steve'),
  32.     gevent.spawn(worker, 'john'),
  33.     gevent.spawn(worker, 'bob'),
  34. ])
  1. Worker steve got task 1
  2. Worker john got task 2
  3. Worker bob got task 3
  4. Worker steve got task 4
  5. Worker john got task 5
  6. Worker bob got task 6
  7. Assigned all work in iteration 1
  8. Worker steve got task 7
  9. Worker john got task 8
  10. Worker bob got task 9
  11. Worker steve got task 10
  12. Worker john got task 11
  13. Worker bob got task 12
  14. Worker steve got task 13
  15. Worker john got task 14
  16. Worker bob got task 15
  17. Worker steve got task 16
  18. Worker john got task 17
  19. Worker bob got task 18
  20. Assigned all work in iteration 2
  21. Worker steve got task 19
  22. Quitting time!
  23. Quitting time!
  24. Quitting time!

分组和池

组是运行中的greenlet的集合，这些greenlet作为组一起管理和调度。它还兼做并行调度程序，借鉴 Python multiprocessing 库。

  1. import gevent
  2. from gevent.pool import Group
  4. def talk(msg):
  5.     for i in xrange(3):
  6.         print(msg)
  8. g1 = gevent.spawn(talk, 'bar')
  9. g2 = gevent.spawn(talk, 'foo')         
  10. g3 = gevent.spawn(talk, 'fizz')  
  12. group = Group()
  13. group.add(g1)                           
  14. group.join()                      # 修改了官方的例子，这里join只会让当前线程等待g1，但g2和g3已经被启动，会被继续安排执行
  1. bar
  2. bar
  3. bar
  4. foo
  5. foo
  6. foo
  7. fizz
  8. fizz
  9. fizz

这对于管理异步任务组非常有用。

如上所述，Group还提供了一个API，用于将作业分派给分组的greenlet并以各种方式收集它们的结果。

  1. import gevent
  2. from gevent import getcurrent
  3. from gevent.pool import Group
  5. group = Group()
  7. def hello_from(n):
  8.     print('Size of group %s' % len(group))
  9.     print('Hello from Greenlet %s' % id(getcurrent()))
  11. group.map(hello_from, xrange(3))
  14. def intensive(n):
  15.     gevent.sleep(3 - n)
  16.     return 'task', n
  18. print('Ordered')
  20. ogroup = Group()
  21. for i in ogroup.imap(intensive, xrange(3)):
  22.     print(i)
  24. print('Unordered')
  26. igroup = Group()
  27. for i in igroup.imap_unordered(intensive, xrange(3)):
  28.     print(i)
  1. Size of group 3
  2. Hello from Greenlet 4340152592
  3. Size of group 3
  4. Hello from Greenlet 4340928912
  5. Size of group 3
  6. Hello from Greenlet 4340928592
  7. Ordered
  8. ('task', 0)
  9. ('task', 1)
  10. ('task', 2)
  11. Unordered
  12. ('task', 2)
  13. ('task', 1)
  14. ('task', 0)

池是一种结构，用于处理需要限制并发的动态数量的greenlets。在需要并行执行许多网络或IO绑定任务的情况下，这通常是可取的。

  1. import gevent
  2. from gevent.pool import Pool
  4. pool = Pool(2)
  6. def hello_from(n):
  7.     print('Size of pool %s' % len(pool))
  9. pool.map(hello_from, xrange(3))
  1. Size of pool 2
  2. Size of pool 2
  3. Size of pool 1

通常在构建gevent驱动的服务时，会将整个服务围绕一个池结构进行中心处理。一个例子可能是在各种套接字（socket）上轮询的类。

  1. from gevent.pool import Pool
  3. class SocketPool(object):
  5.     def __init__(self):
  6.         self.pool = Pool(1000)
  7.         self.pool.start()
  9.     def listen(self, socket):
  10.         while True:
  11.             socket.recv()
  13.     def add_handler(self, socket):
  14.         if self.pool.full():
  15.             raise Exception("At maximum pool size")
  16.         else:
  17.             self.pool.spawn(self.listen, socket)
  19.     def shutdown(self):
  20.         self.pool.kill()

锁和信号量

信号量是一种低级同步原语，它允许greenlet协调和限制并发访问或执行。信号量公开两种方法，获取和释放信号量被获取和释放的次数之差称为信号量的界限。如果信号量范围达到0，它就会阻塞，直到另一个greenlet释放它的捕获。

  1. from gevent import sleep
  2. from gevent.pool import Pool
  3. from gevent.coros import BoundedSemaphore
  5. sem = BoundedSemaphore(2)
  7. def worker1(n):
  8.     sem.acquire()
  9.     print('Worker %i acquired semaphore' % n)
  10.     sleep(0)
  11.     sem.release()
  12.     print('Worker %i released semaphore' % n)
  14. def worker2(n):
  15.     with sem:
  16.         print('Worker %i acquired semaphore' % n)
  17.         sleep(0)
  18.     print('Worker %i released semaphore' % n)
  20. pool = Pool()
  21. pool.map(worker1, xrange(0,2))
  22. pool.map(worker2, xrange(3,6))
  1. Worker 0 acquired semaphore
  2. Worker 1 acquired semaphore
  3. Worker 0 released semaphore
  4. Worker 1 released semaphore
  5. Worker 3 acquired semaphore
  6. Worker 4 acquired semaphore
  7. Worker 3 released semaphore
  8. Worker 4 released semaphore
  9. Worker 5 acquired semaphore
  10. Worker 5 released semaphore

界限为1的信号量称为锁。它提供对一个greenlet的独占执行。它们通常用于确保资源只在程序上下文中使用一次。

翻译持续更新中 …