Gevent ZeroMQ

ZeroMQ 被其作者描述为“充当并发框架的套接字库”。 它是一个非常强大的消息传递层，用于构建并发和分布式应用程序。

ZeroMQ 提供了多种套接字原语，其中最简单的是请求-响应套接字对。 套接字有两种感兴趣的方法 send 和 recv，这两种方法通常都是阻塞操作。 但这已由 Travis Cline 的一个出色的库（现在是 pyzmq 的一部分）进行了补救，该库使用 gevent.socket 以非阻塞方式轮询 ZeroMQ 套接字。

# Note: Remember to ``pip install pyzmq``
import gevent
import zmq.green as zmq

# Global Context
context = zmq.Context()

def server():
    server_socket = context.socket(zmq.REQ)
    server_socket.bind("tcp://127.0.0.1:5000")

    for request in range(1,10):
        server_socket.send("Hello")
        print('Switched to Server for %s' % request)
        # Implicit context switch occurs here
        server_socket.recv()

def client():
    client_socket = context.socket(zmq.REP)
    client_socket.connect("tcp://127.0.0.1:5000")

    for request in range(1,10):

        client_socket.recv()
        print('Switched to Client for %s' % request)
        # Implicit context switch occurs here
        client_socket.send("World")

publisher = gevent.spawn(server)
client    = gevent.spawn(client)

gevent.joinall([publisher, client])

Switched to Server for 1
Switched to Client for 1
Switched to Server for 2
Switched to Client for 2
Switched to Server for 3
Switched to Client for 3
Switched to Server for 4
Switched to Client for 4
Switched to Server for 5
Switched to Client for 5
Switched to Server for 6
Switched to Client for 6
Switched to Server for 7
Switched to Client for 7
Switched to Server for 8
Switched to Client for 8
Switched to Server for 9
Switched to Client for 9

Simple Servers

# On Unix: Access with ``$ nc 127.0.0.1 5000``
# On Window: Access with ``$ telnet 127.0.0.1 5000``

from gevent.server import StreamServer

def handle(socket, address):
    socket.send("Hello from a telnet!\n")
    for i in range(5):
        socket.send(str(i) + '\n')
    socket.close()

server = StreamServer(('127.0.0.1', 5000), handle)
server.serve_forever()

WSGI Servers

Gevent 提供了两个 WSGI 服务器，用于通过 HTTP 提供内容。 此后称为 wsgi 和 pywsgi：

gevent.wsgi.WSGIServer
gevent.pywsgi.WSGIServer
在 1.0.x 之前的早期 gevent 版本中，gevent 使用 libevent 而不是 libev。 Libevent 包含一个快速的 HTTP 服务器，它被 gevent 的 wsgi 服务器使用。

在 gevent 1.0.x 中没有包含 http 服务器。 相反，gevent.wsgi 现在是 gevent.pywsgi 中纯 Python 服务器的别名。

Streaming Servers

如果您使用的是 gevent 1.0.x，则本节不适用

对于那些熟悉流式 HTTP 服务的人来说，核心思想是在标头中我们不指定内容的长度。 相反，我们保持连接打开并沿管道冲洗块，在每个前面加上一个表示块长度的十六进制数字。 当发送大小为零的块时，流将关闭。

HTTP/1.1 200 OK
Content-Type: text/plain
Transfer-Encoding: chunked

8
<p>Hello

9
World</p>

0

无法在 wsgi 中创建上述 HTTP 连接，因为不支持流式传输。 相反，它必须缓冲。

from gevent.wsgi import WSGIServer

def application(environ, start_response):
    status = '200 OK'
    body = '<p>Hello World</p>'

    headers = [
        ('Content-Type', 'text/html')
    ]

    start_response(status, headers)
    return [body]

WSGIServer(('', 8000), application).serve_forever()

然而，使用 pywsgi 我们可以将我们的处理程序编写为生成器并通过 chun 生成结果块

from gevent.pywsgi import WSGIServer

def application(environ, start_response):
    status = '200 OK'

    headers = [
        ('Content-Type', 'text/html')
    ]

    start_response(status, headers)
    yield "<p>Hello"
    yield "World</p>"

WSGIServer(('', 8000), application).serve_forever()

但无论如何，与其他 Python 服务器相比，Gevent 服务器的性能是惊人的。 libev 是一项经过严格审查的技术，众所周知，它的衍生服务器在规模上表现良好。

要进行基准测试，请尝试使用 Apache Benchmark ab 或查看 Python WSGI 服务器的基准以与其他服务器进行比较。

$ ab -n 10000 -c 100 http://127.0.0.1:8000/

Chat Server

最后一个激励示例，一个实时聊天室。 此示例需要 Flask（但不一定如此，您可以使用 Django、Pyramid 等）。 可以在此处找到相应的 Javascript 和 HTML 文件 https://github.com/ZhangTongLe/minichat

# Micro gevent chatroom.
# ----------------------

from flask import Flask, render_template, request

from gevent import queue
from gevent.pywsgi import WSGIServer

import simplejson as json

app = Flask(__name__)
app.debug = True

rooms = {
    'topic1': Room(),
    'topic2': Room(),
}

users = {}

class Room(object):

    def __init__(self):
        self.users = set()
        self.messages = []

    def backlog(self, size=25):
        return self.messages[-size:]

    def subscribe(self, user):
        self.users.add(user)

    def add(self, message):
        for user in self.users:
            print(user)
            user.queue.put_nowait(message)
        self.messages.append(message)

class User(object):

    def __init__(self):
        self.queue = queue.Queue()

@app.route('/')
def choose_name():
    return render_template('choose.html')

@app.route('/<uid>')
def main(uid):
    return render_template('main.html',
        uid=uid,
        rooms=rooms.keys()
    )

@app.route('/<room>/<uid>')
def join(room, uid):
    user = users.get(uid, None)

    if not user:
        users[uid] = user = User()

    active_room = rooms[room]
    active_room.subscribe(user)
    print('subscribe %s %s' % (active_room, user))

    messages = active_room.backlog()

    return render_template('room.html',
        room=room, uid=uid, messages=messages)

@app.route("/put/<room>/<uid>", methods=["POST"])
def put(room, uid):
    user = users[uid]
    room = rooms[room]

    message = request.form['message']
    room.add(':'.join([uid, message]))

    return ''

@app.route("/poll/<uid>", methods=["POST"])
def poll(uid):
    try:
        msg = users[uid].queue.get(timeout=10)
    except queue.Empty:
        msg = []
    return json.dumps(msg)

if __name__ == "__main__":
    http = WSGIServer(('', 5000), app)
    http.serve_forever()

Actors

Actor 模型是由 Erlang 语言推广的更高级别的并发模型。 简而言之，主要思想是您有一组独立的 Actor，它们有一个收件箱，他们可以从中接收来自其他 Actor 的消息。 Actor 内的主循环遍历其消息并根据其所需的行为采取行动。

Gevent 没有原始 Actor 类型，但我们可以使用子类 Greenlet 中的 Queue 非常简单地定义一个。

import gevent
from gevent.queue import Queue

class Actor(gevent.Greenlet):

    def __init__(self):
        self.inbox = Queue()
        Greenlet.__init__(self)

    def receive(self, message):
        """
        Define in your subclass.
        """
        raise NotImplemented()

    def _run(self):
        self.running = True

        while self.running:
            message = self.inbox.get()
            self.receive(message)

In a use case:

import gevent
from gevent.queue import Queue
from gevent import Greenlet

class Pinger(Actor):
    def receive(self, message):
        print(message)
        pong.inbox.put('ping')
        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])

子进程

从 gevent 1.0 开始，已经添加了 gevent.subprocess——Python 的 subprocess 模块的修补版本。 它支持对子进程的协作等待。

# 中间插入子进程
import gevent
from gevent.subprocess import Popen, PIPE

def cron():
    while True:
        print("cron")
        gevent.sleep(0.2)

g = gevent.spawn(cron)
sub = Popen(['sleep 1; uname'], stdout=PIPE, shell=True)
out, err = sub.communicate()
g.kill()
print(out.rstrip())

'''

cron
cron
cron
cron
cron
Linux

'''

许多人还希望将 gevent 和 multiprocessing 一起使用。 最明显的挑战之一是多处理提供的进程间通信在默认情况下是不合作的。 由于基于 multiprocessing.Connection 的对象（例如 Pipe）公开了它们的底层文件描述符，因此 gevent.socket.wait_read 和 wait_write 可用于在实际读取/写入之前协同等待准备读取/准备写入事件：

import gevent
from multiprocessing import Process, Pipe
from gevent.socket import wait_read, wait_write

# To Process
a, b = Pipe()

# From Process
c, d = Pipe()

def relay():
    for i in xrange(10):
        msg = b.recv()
        c.send(msg + " in " + str(i))

def put_msg():
    for i in xrange(10):
        wait_write(a.fileno())
        a.send('hi')

def get_msg():
    for i in xrange(10):
        wait_read(d.fileno())
        print(d.recv())

if __name__ == '__main__':
    proc = Process(target=relay)
    proc.start()

    g1 = gevent.spawn(get_msg)
    g2 = gevent.spawn(put_msg)
    gevent.joinall([g1, g2], timeout=1)

但是请注意，多处理和 gevent 的组合带来了某些依赖于操作系统的陷阱，其中包括：

在符合 POSIX 的系统上分叉后，孩子的 gevent 状态是不适定的。一个副作用是 greenlets 在 multiprocessing.Process 创建之前产生，在父进程和子进程中运行。
上面 put_msg() 中的 a.send() 可能仍会以非合作方式阻塞调用线程：准备写入事件仅确保可以写入一个字节。在尝试写入完成之前，底层缓冲区可能已满。
上述基于 wait_write() / wait_read() 的方法在 Windows 上不起作用（IOError: 3 is not a socket (file is not supported)），因为 Windows 无法监视事件管道。
Python 包 gipc 在符合 POSIX 的系统和 Windows 系统上以一种非常透明的方式为您克服了这些挑战。它提供了 gevent-aware multiprocessing.Process-based child processes 和基于管道的 gevent-cooperative 进程间通信。

锁和信号量

信号量是一种低级同步原语，它允许 greenlet 协调和限制并发访问或执行。信号量公开了两个方法，acquire和release的信号量已经被获取和释放被称为绑定的旗语的次数之间的差异。如果信号量边界达到 0，它将阻塞，直到另一个 greenlet 释放其获取。

from gevent import sleep
from gevent.__semaphore import BoundedSemaphore
from gevent.pool import Pool

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:
        print('Worker %i acquired semaphore' % n)
        sleep(0)
    print('Worker %i released semaphore' % n)

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

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

线程局部变量

Gevent 还允许您指定 greenlet 上下文的本地数据。在内部，这是作为全局查找实现的，它寻址由 greenlet 的getcurrent()值作为键的私有命名空间。

import gevent
from gevent.local import local

stash = local()

def f1():
    stash.x = 1
    print(stash.x)

def f2():
    stash.y = 2
    print(stash.y)
    try:
        stash.x
    except AttributeError:
        print("x is not local to f2")

g1 = gevent.spawn(f1)
g2 = gevent.spawn(f2)

gevent.joinall([g1, g2, g2])

队列

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

例如，如果一个 Greenlet 从队列中抓取一个项目，另一个同时执行的 Greenlet 将不会抓取同一个项目。

import gevent
from gevent.queue import 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 range(1, 25):
        tasks.put_nowait(i)

gevent.spawn(boss).join()

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

'''
输出结果

Worker steve got task 1
Worker john got task 2
Worker nancy got task 3
Worker steve got task 4
Worker john got task 5
Worker nancy got task 6
Worker steve got task 7
Worker john got task 8
Worker nancy got task 9
Worker steve got task 10
Worker john got task 11
Worker nancy got task 12
Worker steve got task 13
Worker john got task 14
Worker nancy got task 15
Worker steve got task 16
Worker john got task 17
Worker nancy got task 18
Worker steve got task 19
Worker john got task 20
Worker nancy got task 21
Worker steve got task 22
Worker john got task 23
Worker nancy got task 24
Quitting time!
Quitting time!
Quitting time!

Process finished with exit code 0

'''

从输出结果中，可以看出从队列取出来的资源，并没有重复。

队列也可以阻塞put或get在需要时阻塞。

每个put and get操作都有一个非阻塞对应物，put_nowait并且 get_nowait不会阻塞，而是在操作不可能时引发gevent.queue.Empty或 引发gevent.queue.Full。

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

import gevent
from gevent.queue import Queue, Empty

tasks = Queue(maxsize=3)  # 队列最大3

def worker(name):
    try:
        while True:
            task = tasks.get(timeout=1)  # 将队列大小减 1
            print('Worker %s got task %s' % (name, task))
            gevent.sleep(0)
    except Empty:
        print('Quitting time!')

def boss():
    """
    老板会等到个别工人下班
     free，因为任务队列的最大大小是 3。
    """

    for i in range(1, 10):
        tasks.put(i)  # 只能添加3次，被阻塞了
        print('我是迭代1')
    print('在迭代 1 中分配所有工作')

    for i in range(10, 20):
        tasks.put(i)
        print('我是迭代2')
    print('在迭代 2 中分配所有工作')

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

'''
我是迭代1
我是迭代1
我是迭代1
Worker steve got task 1
Worker john got task 2
Worker bob got task 3
我是迭代1
我是迭代1
我是迭代1
Worker steve got task 4
Worker john got task 5
Worker bob got task 6
我是迭代1
我是迭代1
我是迭代1
在迭代 1 中分配所有工作
Worker steve got task 7
Worker john got task 8
Worker bob got task 9
我是迭代2
我是迭代2
我是迭代2
Worker steve got task 10
Worker john got task 11
Worker bob got task 12
我是迭代2
我是迭代2
我是迭代2
Worker steve got task 13
Worker john got task 14
Worker bob got task 15
我是迭代2
我是迭代2
我是迭代2
Worker steve got task 16
Worker john got task 17
Worker bob got task 18
我是迭代2
在迭代 2 中分配所有工作
Worker steve got task 19
Quitting time!
Quitting time!
Quitting time!

Process finished with exit code 0

'''

Gevent超时控制

超时是对代码块或 Greenlet 运行时的限制。

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')

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

import gevent
from gevent import Timeout

time_to_wait = 5 # seconds

class TooLong(Exception):
    pass

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

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

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)
thread2 = gevent.spawn(wait)

try:
    thread2.get(timeout=timer)
except Timeout:
    print('Thread 2 timed out')

# --

try:
    gevent.with_timeout(1, wait)
except Timeout:
    print('Thread 3 timed out')

Thread 1 timed out
Thread 2 timed out
Thread 3 timed out

gevent执行状态判断

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) # True
print(loser.started)  # True

# 在 Greenlet 中引发的异常，留在 Greenlet 内。
try:
    gevent.joinall([winner, loser])
except Exception as e:
    print('This will never be reached')

print(winner.value) # 'You win!'
print(loser.value)  # None

print(winner.ready()) # True
print(loser.ready())  # True

print(winner.successful()) # True
print(loser.successful())  # False

# 在失败中引发的异常，不会传播到外面
# greenlet. 堆栈跟踪将打印到标准输出，但它
# 不会展开父级的堆栈。

print(loser.exception)

# 虽然有可能在外面再次引发异常
# raise loser.exception
# or with
# loser.get()

True
True
You win!
None
True
True
True
False
You fail at failing.

杀死僵尸进程

当主程序收到 SIGQUIT 时未能让步的 Greenlet 可能使程序的执行时间比预期的要长。这会导致所谓的“僵尸进程”，需要从 Python 解释器外部杀死它们。

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

import gevent
import signal

def run_forever():
    gevent.sleep(1000)

if __name__ == '__main__':
    gevent.signal(signal.SIGQUIT, gevent.kill)
    thread = gevent.spawn(run_forever)
    thread.join()

案例3：非确定性的task函数

import gevent
import random

def task(pid):
    """
    Some non-deterministic task
    """
    gevent.sleep(random.randint(0, 2) * 0.001)
    print('Task %s done' % pid)

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

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

print('Synchronous:')
synchronous()

print('Asynchronous:')
asynchronous()

'''
输出结果：
Synchronous:
Task 1 done
Task 2 done
Task 3 done
Task 4 done
Task 5 done
Task 6 done
Task 7 done
Task 8 done
Task 9 done
Asynchronous:
Task 0 done
Task 2 done
Task 4 done
Task 6 done
Task 8 done
Task 3 done
Task 9 done
Task 1 done
Task 5 done
Task 7 done

Process finished with exit code 0

'''

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

该程序的重​​要部分是 gevent.spawn将给定函数包装在 Greenlet 线程中的部分。初始化的greenlets列表存储在数组中threads，该数组传递给gevent.joinall阻止当前程序运行所有给定greenlets的函数。只有当所有 greenlet 终止时，执行才会向前推进。

需要注意的重要事实是异步情况下的执行顺序基本上是随机的，并且异步情况下的总执行时间比同步情况少得多。事实上，同步案例完成的最长时间是每个任务暂停 0.002 秒导致整个队列暂停 0.02 秒。在异步情况下，最大运行时间大约为 0.002 秒，因为没有一个任务阻止其他任务的执行。

案例4：从服务器异步获取数据

import gevent.monkey
gevent.monkey.patch_socket()

import gevent
import urllib2
import simplejson as json

def fetch(pid):
    response = urllib2.urlopen('http://json-time.appspot.com/time.json')
    result = response.read()
    json_result = json.loads(result)
    datetime = json_result['datetime']

    print('Process %s: %s' % (pid, datetime))
    return json_result['datetime']

def synchronous():
    for i in range(1,10):
        fetch(i)

def asynchronous():
    threads = []
    for i in range(1,10):
        threads.append(gevent.spawn(fetch, i))
    gevent.joinall(threads)

print('Synchronous:')
synchronous()

print('Asynchronous:')
asynchronous()

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

1、gevent 介绍

小编介绍

gevent是通过协成封装实现的，实现起来让代码更加简洁，用于IO密集型的场景，如果
例如大规模网络并发请求的时候，可以使用Gevent，例如我们经常使用的requests 是Python 程序员经常拿来发出网络请求的模块，他是同步的，为了发出大量并发异步请求，我们可以使用grequest ，他是requests的作者，使用requests模块 结合gevent 构造而成，我们可以直接使用不必重新造轮子。

那另外说一点：而对应的计算密集型的需求，应当采用传统的多线程、多进程方案，就不适合使用gevent。

官方介绍

是一个基于libev的并发库。它为各种并发和网络相关任务提供了一个干净的 API。目标是为您提供使用 gevent 所需的工具，帮助您解决现有的并发问题并立即开始编写异步应用程序。

gevent 中使用的主要模式是Greenlet，这是一个作为 C 扩展模块提供给 Python 的轻量级协程。Greenlets 都在主程序的 OS 进程内运行，但都是协作调度的。这不同于由操作系统提供的任何真正的并行构造 multiprocessing或threading库，它们执行由操作系统调度并且真正并行的自旋进程和 POSIX 线程。

并发的核心思想是将一个更大的任务分解为一组子任务，这些子任务被安排同时或异步运行，而不是一次或同步运行一个。两个子任务之间的切换称为上下文切换。

gevent 中的上下文切换是通过yielding完成的。在这个例子中，我们有两个上下文，它们通过调用 gevent.sleep().

2、Gevent案例：

案例1：

import gevent

def foo():
    print('foo start')
    gevent.sleep(3)
    print('foo end')

def bar():
    print('bar start')
    gevent.sleep(2)
    print('bar end')

def eat():
    print('eat start')
    gevent.sleep(1)
    print('eat end ')

gevent.joinall([
    gevent.spawn(foo),
    gevent.spawn(bar),
    gevent.spawn(eat)
])
#输出结果
'''
foo start
eat start
bar start
eat end 
bar end
foo end

Process finished with exit code 0

'''

通过输出结果可以得出结论，函数执行的顺序默认会按照joinall配置的执行，当函数遇到sleep阻塞的时候，就会执行其他函数，直到最后执行最堵塞的函数完毕。

案例2：

当我们将它用于可以协同安排的网络和IO绑定功能时，gevent的真正力量就来了。
Gevent已经处理了所有细节，以确保您的网络库尽可能隐式地产生greenlet上下文。
我不能强调这是一个强大的习语。但也许一个例子将说明。
在这种情况下，该select()函数通常是一个阻塞调用，它会轮询各种文件描述符。

import time
import gevent
from gevent import select

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

def gr1():
    # Busy waits for a second, but we don't want to stick around...
    print('Started Polling: %s' % tic())
    select.select([], [], [], 2)
    print('Ended Polling: %s' % tic())

def gr2():
    # Busy waits for a second, but we don't want to stick around...
    print('Started Polling: %s' % tic())
    select.select([], [], [], 2)
    print('Ended Polling: %s' % tic())

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

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

'''
运行结果
Started Polling: at 0.0 seconds
Started Polling: at 0.0 seconds
Hey lets do some stuff while the greenlets poll, at 0.0 seconds
Ended Polling: at 2.0 seconds
Ended Polling: at 2.0 seconds
'''