一、背景

大家都知道gevent的機(jī)制是單線程+協(xié)程機(jī)制，當(dāng)遇到可能會(huì)阻塞的操作時(shí)，就切換到可運(yùn)行的協(xié)程中繼續(xù)運(yùn)行，以此來(lái)實(shí)現(xiàn)提交系統(tǒng)運(yùn)行效率的目標(biāo)，但是具體是怎么實(shí)現(xiàn)的呢？讓我們直接從代碼中看一下吧。

二、切換機(jī)制

讓我們從socket的send、recv方法入手：

def recv(self, *args): while 1: try: return self._sock.recv(*args) except error as ex: if ex.args[0] != EWOULDBLOCK or self.timeout == 0.0:raise # QQQ without clearing exc_info test__refcount.test_clean_exit fails sys.exc_clear() self._wait(self._read_event)

這里會(huì)開(kāi)啟一個(gè)死循環(huán)，在循環(huán)中調(diào)用self._sock.recv()方法，并捕獲異常，當(dāng)錯(cuò)誤是EWOULDBLOCK時(shí)，則調(diào)用self._wait(self._read_event)方法，該方法其實(shí)是：_wait = _wait_on_socket，_wait_on_socket方法的定義在文件：_hub_primitives.py中，如下：

# Suitable to be bound as an instance methoddef wait_on_socket(socket, watcher, timeout_exc=None): if socket is None or watcher is None: # test__hub TestCloseSocketWhilePolling, on Python 2; Python 3 # catches the EBADF differently. raise ConcurrentObjectUseError('The socket has already been closed by another greenlet') _primitive_wait(watcher, socket.timeout, timeout_exc if timeout_exc is not None else _NONE, socket.hub)

該方法其實(shí)是調(diào)用了函數(shù)：_primitive_wait()，其仍然在文件：_hub_primitives.py中定義，如下：

def _primitive_wait(watcher, timeout, timeout_exc, hub): if watcher.callback is not None: raise ConcurrentObjectUseError(’This socket is already used by another greenlet: %r’ % (watcher.callback, )) if hub is None: hub = get_hub() if timeout is None: hub.wait(watcher) return timeout = Timeout._start_new_or_dummy( timeout, (timeout_exc if timeout_exc is not _NONE or timeout is None else _timeout_error(’timed out’))) with timeout: hub.wait(watcher)

這里其實(shí)是調(diào)用了hub.wait()函數(shù)，該函數(shù)的定義在文件_hub.py中，如下：

class WaitOperationsGreenlet(SwitchOutGreenletWithLoop): # pylint:disable=undefined-variable def wait(self, watcher): ''' Wait until the *watcher* (which must not be started) is ready. The current greenlet will be unscheduled during this time. ''' waiter = Waiter(self) # pylint:disable=undefined-variable watcher.start(waiter.switch, waiter) try: result = waiter.get() if result is not waiter:raise InvalidSwitchError( ’Invalid switch into %s: got %r (expected %r; waiting on %r with %r)’ % ( getcurrent(), # pylint:disable=undefined-variable result, waiter, self, watcher )) finally: watcher.stop()

watcher.stop()

該類(lèi)WaitOperationsGreenlet是Hub的基類(lèi)，其方法wait中的邏輯是：生成一個(gè)Waiter對(duì)象，并調(diào)用watcher.start(waiter.switch, waiter)方法，watcher是最開(kāi)始recv方法中使用的self._read_event，watcher是gevent的底層事件框架libev中的概念；同時(shí)還有一個(gè)waiter對(duì)象，它類(lèi)似與python中的future概念，該對(duì)象有一個(gè)switch()方法以及get()方法，當(dāng)沒(méi)有得到結(jié)果沒(méi)有準(zhǔn)備好時(shí)，調(diào)用waiter.get()方法回導(dǎo)致協(xié)程被掛起；get()函數(shù)的定義如下：

def get(self): '''If a value/an exception is stored, return/raise it. Otherwise until switch() or throw() is called.''' if self._exception is not _NONE: if self._exception is None: return self.value getcurrent().throw(*self._exception) # pylint:disable=undefined-variable else: if self.greenlet is not None: raise ConcurrentObjectUseError(’This Waiter is already used by %r’ % (self.greenlet, )) self.greenlet = getcurrent() # pylint:disable=undefined-variable try: return self.hub.switch() finally: self.greenlet = None

在get()中最關(guān)鍵的是self.hub.switch()函數(shù)，該函數(shù)將執(zhí)行權(quán)轉(zhuǎn)移到hub，并繼續(xù)運(yùn)行，至此已經(jīng)分析完了當(dāng)在worker協(xié)程中從網(wǎng)絡(luò)獲取數(shù)據(jù)遇到阻塞時(shí)，如何避免阻塞并切換到hub中的實(shí)現(xiàn)，至于何時(shí)再切換會(huì)worker協(xié)程，我們后續(xù)再繼續(xù)分析。

總結(jié)

要記得gevent中一個(gè)重要的概念，協(xié)程切換不是調(diào)用而是執(zhí)行權(quán)的轉(zhuǎn)移，從可能會(huì)阻塞的協(xié)程切換到hub，并由hub在合適的時(shí)機(jī)切換到另一個(gè)可以繼續(xù)運(yùn)行的協(xié)程繼續(xù)執(zhí)行；gevent通過(guò)這種形式實(shí)現(xiàn)了提高io密集型應(yīng)用吞吐率的目標(biāo)。

以上就是本文的全部?jī)?nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持好吧啦網(wǎng)。