关于gevent的一些理解(一)

def foo():
    print('Running in foo')
    gevent.sleep(0) #让当前的greenlet睡眠N秒,这0标识控制其它协程而不会让其它进程睡眠
    print('Explicit context switch to foo again')

def bar():
    print('Explicit context to bar')
    gevent.sleep(0)
    print('Implicit context switch back to bar')

gevent.joinall([  #gevent.Greenlet实例,直到这个greenlet完成或者超时
    gevent.spawn(foo),  #spawn可以实现一个grennlet实例并且加到队列并且启动,效果类似于gevent.Greenlet(foo).start()
    gevent.spawn(bar),
])

import gevent
from gevent import select #类似于内置的select.select()实现(请关注http://www.dongwm.com/archives/guanyuselectyanjiu/),只是将线程操作改成了greenlet

start = time.time()
tic = lambda: 'at %1.1f seconds' % (time.time() - start)

def gr1():
    print('Started Polling: ', tic())
    select.select([], [], [], 2)  #参数分别是,等待的可读列表,等待的可写列表,等待的可执行列表,超时时间(这里是2秒)
    print('Ended Polling: ', tic())

def gr2():
    print('Started Polling: ', tic())
    select.select([], [], [], 2)
    print('Ended Polling: ', tic())

def gr3():
    print("Hey lets do some stuff while the greenlets poll, at", tic())
    gevent.sleep(1)

gevent.joinall([
    gevent.spawn(gr1),
    gevent.spawn(gr2),
    gevent.spawn(gr3),
])

import gevent
import random

def task(pid):
    gevent.sleep(random.randint(0,2)*0.001)
    print('Task', pid, 'done')

def synchronous():  #同步
    for i in range(1,10):
        task(i)

def asynchronous(): #异步
    threads = [gevent.spawn(task, i) for i in xrange(10)]
    gevent.joinall(threads)

print('Synchronous:')
synchronous()

print('Asynchronous:')
asynchronous()

import gevent
from gevent import Greenlet

def foo(message, n):
    gevent.sleep(n)
    print(message)

thread1 = Greenlet.spawn(foo, "Hello", 1)  #实例化Greenlet
thread2 = gevent.spawn(foo, "I live!", 2) #实例化gevent,其实也是创建Greenlet实例,只是包装了一下
thread3 = gevent.spawn(lambda x: (x+1), 2)  #一个lambda表达式

threads = [thread1, thread2, thread3]
gevent.joinall(threads) #等待所有greenlet完成

import gevent
from gevent import Greenlet

class MyGreenlet(Greenlet):  #重载Greenlet类

    def __init__(self, message, n):
        Greenlet.__init__(self)
        self.message = message
        self.n = n

    def _run(self): #重写_run方法
        print(self.message)
        gevent.sleep(self.n)

g = MyGreenlet("Hi there!", 3)
g.start()
g.join()

 

import gevent

def win():
    return 'You win!'

def fail():
    raise Exception('You fail at failing.')

winner = gevent.spawn(win)
loser = gevent.spawn(fail)

print(winner.started) # started表示的Greenlet是否已经开始,返回布尔值
print(loser.started)  # True

try:
    gevent.joinall([winner, loser])
except Exception as e:
    print('This will never be reached')

print(winner.value) # value表示greenlet实例返回值:'You win!'
print(loser.value)  # None

print(winner.ready()) # 是否已停止Greenlet的布尔值,True
print(loser.ready())  # True

print(winner.successful()) # 表示的Greenlet是否已成功停止，而不是抛出异常,True
print(loser.successful())  # False
print(loser.exception) #打印异常的报错信息

import gevent
from gevent import Timeout

seconds = 10

timeout = Timeout(seconds)
timeout.start()

def wait():
    gevent.sleep(10)

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

import gevent
from gevent import Timeout

time_to_wait = 5 # seconds

class TooLong(Exception):
    pass

with Timeout(time_to_wait, TooLong):
    gevent.sleep(10)

import gevent
from gevent import Timeout

def wait():
    gevent.sleep(2)

timer = Timeout(1).start()
thread1 = gevent.spawn(wait)  #这种超时类型前面讲过

try:
    thread1.join(timeout=timer)
except Timeout:
    print('Thread 1 timed out')

timer = Timeout.start_new(1) #start_new是一个快捷方式
thread2 = gevent.spawn(wait)

try:
    thread2.get(timeout=timer) #get返回greenlet的结果,包含异常
except Timeout:
    print('Thread 2 timed out')

try:
    gevent.with_timeout(1, wait) #如果超时前返回异常,取消这个方法
except Timeout:
    print('Thread 3 timed out')

import gevent
from gevent.event import AsyncResult

a = AsyncResult() #保存一个值或者一个异常的事件实例

def setter():
    gevent.sleep(3)  #3秒后唤起所有线程的a的值
    a.set() #保存值,唤起等待线程

def waiter():
    a.get() # 3秒后get方法不再阻塞,返回存贮的值或者异常
    print 'I live!'

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

import gevent

from gevent.event import AsyncResult
a = AsyncResult()

def setter():
    gevent.sleep(3)
    a.set('Hello!')

def waiter():
    print a.get()

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

import gevent
from gevent.queue import Queue  #类似于内置的Queue

tasks = Queue() #队列实例

def worker(n):
    while not tasks.empty():
        task = tasks.get()
        print('Worker %s got task %s' % (n, task))
        gevent.sleep(0)

    print('Quitting time!')

def boss():
    for i in xrange(1,25):
        tasks.put_nowait(i) #非阻塞的把数据放到队列里面

gevent.spawn(boss).join()

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

import gevent
from gevent.queue import Queue, Empty

tasks = Queue(maxsize=3)  #限制队列的长度

def worker(n):
    try:
        while True:
            task = tasks.get(timeout=1) # 减少队列,超时为1秒
            print('Worker %s got task %s' % (n, task))
            gevent.sleep(0)
    except Empty:
        print('Quitting time!')

def boss():
    """
    Boss will wait to hand out work until a individual worker is
    free since the maxsize of the task queue is 3.
    """

    for i in xrange(1,10):
        tasks.put(i)  #这里boss没有盲目的不停放入数据,而是在当最大三个队列数有空余才放入数据,事实上方法转换过程中,boss放入三个数据,worker取出三个数据,boss再放入数据....
    print('Assigned all work in iteration 1')

    for i in xrange(10,20):
        tasks.put(i)
    print('Assigned all work in iteration 2')

gevent.joinall([
    gevent.spawn(boss),
    gevent.spawn(worker, 'steve'),
    gevent.spawn(worker, 'john'),
    gevent.spawn(worker, 'bob'),
])

from gevent.pool import Group
def talk(msg):
    for i in xrange(3):
        print(msg)

g1 = gevent.spawn(talk, 'bar')
g2 = gevent.spawn(talk, 'foo')
g3 = gevent.spawn(talk, 'fizz')

group = Group() #保持greenlet实例的组运行,连接到没个项目,在其完成后删除
group.add(g1)
group.add(g2)
group.join()

group.add(g3)
group.join()

import gevent
from gevent import getcurrent
from gevent.pool import Group

group = Group()

def hello_from(n):
    print('Size of group', len(group))
    print('Hello from Greenlet %s' % id(getcurrent()))  #获取当前gevent实例的id

group.map(hello_from, xrange(3)) #map迭代方法,参数为方法和其参数

def intensive(n):
    gevent.sleep(3 - n)
    return 'task', n

print('Ordered')

ogroup = Group()
for i in ogroup.imap(intensive, xrange(3)):  #相当于 itertools.imap,返回一个迭代器, 它是调用了一个其值在输入迭代器上的函数, 返回结果. 它类似于函数 map() , 只是前者在
#任意输入迭代器结束后就停止(而不是插入None值来补全所有的输入)
    print(i)

print('Unordered')

igroup = Group()
for i in igroup.imap_unordered(intensive, xrange(3)):
    print(i)

import gevent
from gevent import getcurrent
from gevent.pool import Pool

pool = Pool(2)

def hello_from(n):
    print('Size of pool', len(pool))

pool.map(hello_from, xrange(3))

from gevent.pool import Pool

class SocketPool(object):

    def __init__(self):
        self.pool = Pool(1000)  #设置池容量1000
        self.pool.start()

    def listen(self, socket):
        while True:
            socket.recv()

    def add_handler(self, socket):
        if self.pool.full(): #容量慢报错
            raise Exception("At maximum pool size")
        else: #否则执行在新的grenlet里面执行listen方法
            self.pool.spawn(self.listen, socket)

    def shutdown(self):
        self.pool.kill() #关闭pool

from gevent import sleep

from gevent.pool import Pool
from gevent.coros import BoundedSemaphore

sem = BoundedSemaphore(2) #设定对共享资源的访问数量

def worker1(n):
    sem.acquire() #获取资源
    print('Worker %i acquired semaphore' % n)
    sleep(0)
    sem.release()  #释放资源
    print('Worker %i released semaphore' % n)

def worker2(n):
    with sem: #使用with关键字
        print('Worker %i acquired semaphore' % n)
        sleep(0)
    print('Worker %i released semaphore' % n)

pool = Pool()
pool.map(worker1, xrange(0,2))
pool.map(worker2, xrange(3,6))

from gevent.queue import Queue
from gevent import Greenlet

class Actor(gevent.Greenlet): #自定义actor类

    def __init__(self):
        self.inbox = Queue() #收件箱作为一个队列
        Greenlet.__init__(self)

    def receive(self, message):
        raise NotImplemented() #内置常量,表面意为没有实施

    def _run(self): #
        self.running = True

        while self.running:
            message = self.inbox.get() #获取队列数据
            self.receive(message)

class Pinger(Actor):
    def receive(self, message): #重写方法
        print message
        pong.inbox.put('ping') #当获取收件箱有数据,获取数据,再放入数据(注意:是ping中放pong数据),其中pong是一个局部变量,它是Ponger的实例,以下的同理
        gevent.sleep(0)

class Ponger(Actor):
    def receive(self, message):
        print message
        ping.inbox.put('pong')
        gevent.sleep(0)

ping = Pinger()
pong = Ponger()

ping.start()
pong.start()

ping.inbox.put('start') #最开始都是阻塞的,给一个触发
gevent.joinall([ping, pong])