Processing math: 100%

实用python工具包锦集

sh

sh 是一个成熟，用于替代 subprocess，它允许你调用任何程序，就像它是一个函数，支持 Python2.6 - 3.5

1
2
from sh import ifconfig
print ifconfig("eth0")

mycli

mycli 是一个带语法高亮、自动补全的 MySQL 命令行客户端工具。例如，连接数据库方法：

1
mycli -h localhost -u 用户名 数据库

python-fire

Fire 是 Google 开源的 Python 库，可自动将您的代码转变成 CLI，无需您做任何额外工作。您不必定义参数，设置帮助信息，或者编写定义代码运行方式的 main 函数。相反，您只需从 main 模块调用“Fire”函数，其余工作全部交由 Python Fire 来完成。示例代码如下：

1
2
3
4
5
6
7
8
9
10
11

import fire
class Example(object):
    def hello(self, name='world'):
        """Says hello to the specified name."""
        return 'Hello {name}!'.format(name=name)

def main():
    fire.Fire(Example)
if __name__ == '__main__':
    main()
1
2
3
4
5
6
7
# 在终端中调用效果如下：
$ ./example.py hello
Hello world!
$ ./example.py hello David
Hello David!
$ ./example.py hello --name=Google
Hello Google!

pdir2

Python 程序员需要一个更好的 dir() —— 以更加友好的显示 dir() 输出的结果

freezegun

时间漫步模块，模拟到某一个时间，使用简单方式多样，实现了装饰器、上下文等调用方式。示例代码如下：

1
2
3
4
5
6
7
8
from freezegun import freeze_time
import datetime
import unittest


@freeze_time("2012-01-14")
def test():
    assert datetime.datetime.now() == datetime.datetime(2012, 1, 14)

records

Kenneth Reitz 大神的for Humans™系列，Records 是一个支持大多数主流关系数据库的原生 SQL 查询第三方库。API 友好，使用简单、支持命令行模式、功能多样。与此同时该库只有 500 行代码，可以当作入门阅读源码的项目，同时学习大神的编程技巧与习惯，示例代码如下：

pygorithm

一个帮助学习主要算法的库，可以通过理解这些算法的实现，提高自己的算法水平

huey

结合 redis 实现的轻量任务队列，但是支持功能还是很多的：

  • 多进程、多线程、协程
  • 任务定时执行
  • 任务执行失败重试
  • 结果存储

joblib

使用 Python 方便的进行并行计算，示例代码如下：

1
2
3
from joblib import Parallel, delayed
from math import sqrt
Parallel(n_jobs=1)(delayed(sqrt)(i**2) for i in range(10))

thefuck

在 Linux 命令行中，当你输入的命令有错误后，直接输入 fuck 就可以自动执行修复后的命令

pook

模拟 HTTP 请求结果的库，可用于单元测试等场景。采用装饰器方式调用的示例代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14

import pook
import requests

@pook.get('http://httpbin.org/status/500', reply=204)
@pook.get('http://httpbin.org/status/400', reply=200)
def fetch(url):
    return requests.get(url)

res = fetch('http://httpbin.org/status/400')
print('#1 status:', res.status_code)

res = fetch('http://httpbin.org/status/500')
print('#2 status:', res.status_code)

tenacity

使用该库可以优雅地实现各种需求的重试。示例代码如下：

1
2
3
4
5
6
7
from tenacity import retry, stop_after_attempt

# 通过装饰器，实现遇到异常重试3次
@retry(stop=stop_after_attempt(3)) 
def get_data(url):
    response = requests.get(url)
    response_json = response.json()

pudb

基于控制台的全屏 Python 可视化调试器。比 pdb 好用太多了，特性：

  • 源码语法高亮，栈、断点、变量可见并且一直动态更新。变量展示还有很多可以定制化的功能。
  • 基于键盘，简单高效。支持 VI 的鼠标移动。还支持 PDB 的某些命令
  • 支持查找源代码，可以使用 m 代用 module browser 查看载入的模块
  • 断点设置。鼠标移到某行代码，按 b，然后可以在断点窗口编辑断点

loguru

一个让 Python 记录日志变得简单的库

click

Python 的第三方库，用于快速创建命令行。支持装饰器方式调用、多种参数类型、自动生成帮助信息等。示例代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
import click

@click.command()
@click.option("--count", default=1, help="Number of greetings.")
@click.option("--name", prompt="Your name",
              help="The person to greet.")
def hello(count, name):
    """Simple program that greets NAME for a total of COUNT times."""
    for _ in range(count):
        click.echo("Hello, %s!" % name)

if __name__ == '__main__':
    hello()
1
2
3
4
5
6
7

# 下面为运行效果
$ python hello.py --count=3
Your name: Click
Hello, Click!
Hello, Click!
Hello, Click!

pyright

微软出品的 Python 静态类型检查工具。执行速度快，适合大型 Python 项目，引用一句话：动态语言一时爽，重构火葬场

PySnooper

Python 的第三方调试库。让你通过装饰器方法，方便的知道每一行程序运行后的结果，而不需要再手动增加 print 展示过程数据、调试程序。示例代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
import pysnooper

@pysnooper.snoop()
def number_to_bits(number):
    if number:
        bits = []
        while number:
            number, remainder = divmod(number, 2)
            bits.insert(0, remainder)
        return bits
    else:
        return [0]
        
number_to_bits(6)
1
2
3
4
5
6
7
8
9
10
11
# 输出如下
Starting var:.. number = 6
15:29:11.327032 call         4 def number_to_bits(number):
15:29:11.327032 line         5     if number:
15:29:11.327032 line         6         bits = []
New var:....... bits = []
15:29:11.327032 line         7         while number:
15:29:11.327032 line         8             number, remainder = divmod(number, 2)
New var:....... remainder = 0
Modified var:.. number = 3
....

装饰器妙用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
import datetime


def count_time(func):
    def int_time(*args, **kwargs):
        start_time = datetime.datetime.now()  # 程序开始时间
        func()
        over_time = datetime.datetime.now()   # 程序结束时间
        total_time = (over_time-start_time).total_seconds()
        print('程序共计%s秒' % total_time)
    return int_time


@count_time
def main():
    print('>>>>开始计算函数运行时间')
    for i in range(1, 1000):　　　　　　# 可以是任意函数  ， 这里故意模拟函数的运行时间
        for j in range(i):
            print(j)


if __name__ == '__main__':
    main()

arrow

还在为处理时间、时区、转化、夏令时等问题而头疼吗？这个 Python 的第三方时间库。提供了更便捷的方式来创建、操作和格式化时间和日期，用更少的代码来处理时间和日期

展开全文 >>

Celery4.1 中文文档

Application(应用)

Celery 库在使用之前必须初始化，一个celery实例被称为一个应用（或者缩写 app）。

Celery 应用是线程安全的，所以多个不同配置、不同组件、不同任务的 应用可以在一个进程空间里共存。
下面创建一个 celery 应用：

1
2
3
4
>>> from celery import Celery
>>> app = Celery()
>>> app
<Celery __main__:0x100469fd0>

最后一行显示的是 celery 应用的文本表示： 包含应用类的名称（Celery），当前主模块的名称（main），以及应用对象的内存地址（0x100469fd0）。

Main Name

上述文本表示中只有一部分是重要的，那就是主模块名称。下面分析下它为何重要。

当你发送一个消息给 Celery，消息中不会包含任何源码，而只有你想要执行的任务的名称。这就好像因特网上的域名映射原理一般：每个执行单元维护着一个任务名称到实际任务函数的映射，这个映射被称为任务注册表。

1
2
3
4
5
6
7
8
9
10
11
12
>>> @app.task
... def add(x, y):
...     return x + y

>>> add
<@task: __main__.add>

>>> add.name
__main__.add

>>> app.tasks['__main__.add']
<@task: __main__.add>

现在，你又看到显示 main 模块名称；当Celery不能探查到这个任务函数属于哪个模块时，它将使用主模块名称来产生任务名称的前缀。

这在有些情况下会产生问题：

  1. 定义任务的主模块作为一个程序运行。
  2. 应用在python交互终端创建。

例如下面程序，定义任务的模块还调用 app.worker_main() 来启动一个工作单元：
task.py

1
2
3
4
5
6
7
8
from celery import Celery
app = Celery()

@app.task
def add(x, y): return x + y

if __name__ == '__main__':
    app.worker_main()

当这个模块运行，任务将以前缀__main__ 命名，但是当该模块被其他进程引入来运行一个任务，这个任务的名称将以前缀 tasks 命名（即这个模块的真实名称）。

1
2
3
>>> from tasks import add
>>> add.name
tasks.add

你可以给主模块声明另外一个名称

1
2
3
4
5
6
7
8
9
10
>>> app = Celery('tasks')
>>> app.main
'tasks'

>>> @app.task
... def add(x, y):
...     return x + y

>>> add.name
tasks.add

具体可以查看名称这节

展开全文 >>

Gevent 简明教程

前述

进程 线程 协程 异步

并发编程（不是并行）目前有四种方式：多进程、多线程、协程和异步。

  • 多进程编程在python中有类似C的os.fork,更高层封装的有multiprocessing标准库
  • 多线程编程python中有Thread和threading
  • 异步编程在linux下主+要有三种实现select，poll，epoll

  • 协程在python中通常会说到yield，关于协程的库主要有greenlet,stackless,gevent,eventlet等实现。

    进程

  • 不共享任何状态

  • 调度由操作系统完成
  • 有独立的内存空间（上下文切换的时候需要保存栈、cpu寄存器、虚拟内存、以及打开的相关句柄等信息，开销大）

  • 通讯主要通过信号传递的方式来实现（实现方式有多种，信号量、管道、事件等，通讯都需要过内核，效率低）

    线程

  • 调度完全由用户控制

  • 一个线程（进程）可以有多个协程
  • 每个线程（进程）循环按照指定的任务清单顺序完成不同的任务（当任务被堵塞时，执行下一个任务；当恢复时，再回来执行这个任务；任务间切换只需要保存任务的上下文，没有内核的开销，可以不加锁的访问全局变量）
  • 协程需要保证是非堵塞的且没有相互依赖
  • 协程基本上不能同步通讯，多采用异步的消息通讯，效率比较高

总结

  • 进程拥有自己独立的堆和栈，既不共享堆，亦不共享栈，进程由操作系统调度
  • 线程拥有自己独立的栈和共享的堆，共享堆，不共享栈，线程亦由操作系统调度(标准线程是的)
  • 协程和线程一样共享堆，不共享栈，协程由程序员在协程的代码里显示调度

聊聊协程

协程，又称微线程，纤程。

Python的线程并不是标准线程，是系统级进程，线程间上下文切换有开销，而且Python在执行多线程时默认加了一个全局解释器锁（GIL），因此Python的多线程其实是串行的，所以并不能利用多核的优势，也就是说一个进程内的多个线程只能使用一个CPU。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
def coroutine(func):
    def ret():
        f = func()
        f.next()
        return f
    return ret


@coroutine
def consumer():
    print "Wait to getting a task"
    while True:
        n = (yield)
        print "Got %s",n


import time
def producer():
    c = consumer()
    task_id = 0
    while True:
        time.sleep(1)
        print "Send a task to consumer" % task_id
        c.send("task %s" % task_id)

if __name__ == "__main__":
    producer()

结果：

1
2
3
4
5
6
7
8
Wait to getting a task
Send a task 0 to consumer
Got task 0
Send a task 1 to consumer
Got task 1
Send a task 2 to consumer
Got task 2
...

传统的生产者-消费者模型是一个线程写消息，一个线程取消息，通过锁机制控制队列和等待，但容易死锁。
如果改用协程，生产者生产消息后，直接通过yield跳转到消费者开始执行，待消费者执行完毕后，切换回生产者继续生产，效率极高。

Gevent

介绍

gevent是基于协程的Python网络库。特点：

  • 基于libev的快速事件循环(Linux上epoll，FreeBSD上kqueue）。
  • 基于greenlet的轻量级执行单元。
  • API的概念和Python标准库一致(如事件，队列)。
  • 可以配合socket，ssl模块使用。
  • 能够使用标准库和第三方模块创建标准的阻塞套接字(gevent.monkey)。
  • 默认通过线程池进行DNS查询,也可通过c-are(通过GEVENT_RESOLVER=ares环境变量开启）。
  • TCP/UDP/HTTP服务器
  • 子进程支持（通过gevent.subprocess）
  • 线程池

安装和依赖

依赖于greenlet library

支持python 2.6+ 、3.3+

核心部分

  • Greenlets
  • 同步和异步执行
  • 确定性
  • 创建Greenlets
  • Greenlet状态
  • 程序停止
  • 超时
  • 猴子补丁

Greenlets

gevent中的主要模式, 它是以C扩展模块形式接入Python的轻量级协程。 全部运行在主程序操作系统进程的内部，但它们被程序员协作式地调度。

在任何时刻，只有一个协程在运行。

区别于multiprocessing、threading等提供真正并行构造的库， 这些库轮转使用操作系统调度的进程和线程，是真正的并行

同步和异步执行

并发的核心思想在于，大的任务可以分解成一系列的子任务，后者可以被调度成 同时执行或异步执行，而不是一次一个地或者同步地执行。两个子任务之间的 切换也就是上下文切换。

在gevent里面，上下文切换是通过yielding来完成的.

1
2
3
4
5
6
7
8
9
10
11
12
13
import gevent
def foo():
    print('Running in foo')
    gevent.sleep(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.spawn(foo),
    gevent.spawn(bar),
])

执行结果：

1
2
3
4
Running in foo
Explicit context to bar
Explicit context switch to foo again
Implicit context switch back to bar

网络延迟或IO阻塞隐式交出greenlet上下文的执行权。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
import time
import gevent
from gevent import select
start = time.time()
tic = lambda: 'at %1.1f seconds' % (time.time() - start)
def gr1():
    print('Started Polling: %s' % tic())
    select.select([], [], [], 1)
    print('Ended Polling: %s' % tic())
def gr2():
    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),
])

执行结果

1
2
3
4
5
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 1.0 seconds
Ended Polling: at 2.0 seconds

同步VS异步

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

import gevent
import random
def task(pid):
    gevent.sleep(random.randint(0,2)*0.001)
    print('Task %s done' % pid)
def synchronous():
    for i in xrange(5):
        task(i)
def asynchronous():
    threads = [gevent.spawn(task, i) for i in xrange(5)]
    gevent.joinall(threads)
    print('Synchronous:')
    synchronous()
    print('Asynchronous:')
    asynchronous()

执行结果

1
2
3
4
5
6
7
8
9
10
11
12
Synchronous:
Task 0 done
Task 1 done
Task 2 done
Task 3 done
Task 4 done
Asynchronous:
Task 2 done
Task 0 done
Task 1 done
Task 3 done
Task 4 done

确定性

greenlet具有确定性。在相同配置相同输入的情况下，它们总是会产生相同的输出。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
import time
def echo(i):
    time.sleep(0.001)
    return i
# Non Deterministic Process Pool
from multiprocessing.pool import Pool
p = Pool(10)
run1 = [a for a in p.imap_unordered(echo, xrange(10))]
run2 = [a for a in p.imap_unordered(echo, xrange(10))]
run3 = [a for a in p.imap_unordered(echo, xrange(10))]
run4 = [a for a in p.imap_unordered(echo, xrange(10))]
print(run1 == run2 == run3 == run4)
# Deterministic Gevent Pool
from gevent.pool import Pool
p = Pool(10)
run1 = [a for a in p.imap_unordered(echo, xrange(10))]
run2 = [a for a in p.imap_unordered(echo, xrange(10))]
run3 = [a for a in p.imap_unordered(echo, xrange(10))]
run4 = [a for a in p.imap_unordered(echo, xrange(10))]
print(run1 == run2 == run3 == run4)

执行结果：

1
2
False
True

即使gevent通常带有确定性，当开始与如socket或文件等外部服务交互时， 不确定性也可能溜进你的程序中。因此尽管gevent线程是一种“确定的并发”形式， 使用它仍然可能会遇到像使用POSIX线程或进程时遇到的那些问题。

涉及并发长期存在的问题就是竞争条件(race condition)(当两个并发线程/进程都依赖于某个共享资源同时都尝试去修改它的时候， 就会出现竞争条件),这会导致资源修改的结果状态依赖于时间和执行顺序。 这个问题，会导致整个程序行为变得不确定。

解决办法: 始终避免所有全局的状态.

创建Greenlets

gevent对Greenlet初始化提供了一些封装.

1
2
3
4
5
6
7
8
9
10
import gevent
from gevent import Greenlet
def foo(message, n):
    gevent.sleep(n)
    print(message)
    thread1 = Greenlet.spawn(foo, "Hello", 1)
    thread2 = gevent.spawn(foo, "I live!", 2)
    thread3 = gevent.spawn(lambda x: (x+1), 2)
    threads = [thread1, thread2, thread3]
    gevent.joinall(threads)

执行结果：

1
2
Hello
I live!

除使用基本的Greenlet类之外，你也可以子类化Greenlet类，重载它的_run方法。

1
2
3
4
5
6
7
8
9
10
11
12
13
14

import gevent
from gevent import Greenlet
class MyGreenlet(Greenlet):
    def __init__(self, message, n):
        Greenlet.__init__(self)
        self.message = message
        self.n = n
    def _run(self):
        print(self.message)
        gevent.sleep(self.n)
g = MyGreenlet("Hi there!", 3)
g.start()
g.join()

执行结果：

1
Hi there!

Greenlet状态

greenlet的状态通常是一个依赖于时间的参数：

  • started – Boolean, 指示此Greenlet是否已经启动
  • ready() – Boolean, 指示此Greenlet是否已经停止
  • successful() – Boolean, 指示此Greenlet是否已经停止而且没抛异常
  • value – 任意值, 此Greenlet代码返回的值
  • exception – 异常, 此Greenlet内抛出的未捕获异常

程序停止

程序
当主程序(main program)收到一个SIGQUIT信号时，不能成功做yield操作的 Greenlet可能会令意外地挂起程序的执行。这导致了所谓的僵尸进程， 它需要在Python解释器之外被kill掉。

通用的处理模式就是在主程序中监听SIGQUIT信号，调用gevent.shutdown退出程序。

1
2
3
4
5
6
7
8
9

import gevent
import signal
def run_forever():
    gevent.sleep(1000)
    if __name__ == '__main__':
        gevent.signal(signal.SIGQUIT, gevent.shutdown)
        thread = gevent.spawn(run_forever)
        thread.join()

超时

通过超时可以对代码块儿或一个Greenlet的运行时间进行约束。

1
2
3
4
5
6
7
8
9
10
11
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')

超时类

1
2
3
4
5
6
7
8
import gevent
from gevent import Timeout
time_to_wait = 5 # seconds
    class TooLong(Exception):
        pass

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

另外，对各种Greenlet和数据结构相关的调用，gevent也提供了超时参数。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

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

执行结果：

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

猴子补丁(Monkey patching)

1
2
3
4
5
6
7
8
9
10
11
import socket
print(socket.socket)
print("After monkey patch")
from gevent import monkey
monkey.patch_socket()
print(socket.socket)
import select
print(select.select)
monkey.patch_select()
print("After monkey patch")
print(select.select)

执行结果：

1
2
3
4
5
6
class 'socket.socket'
After monkey patch
class 'gevent.socket.socket'
built-in function select
After monkey patch
function select at 0x1924de8

Python的运行环境允许我们在运行时修改大部分的对象，包括模块，类甚至函数。 这是个一般说来令人惊奇的坏主意，因为它创造了“隐式的副作用”，如果出现问题 它很多时候是极难调试的。虽然如此，在极端情况下当一个库需要修改Python本身 的基础行为的时候，猴子补丁就派上用场了。在这种情况下，gevent能够修改标准库里面大部分的阻塞式系统调用，包括socket、ssl、threading和 select等模块，而变为协作式运行。

例如，Redis的python绑定一般使用常规的tcp socket来与redis-server实例通信。 通过简单地调用gevent.monkey.patch_all()，可以使得redis的绑定协作式的调度 请求，与gevent栈的其它部分一起工作。

这让我们可以将一般不能与gevent共同工作的库结合起来，而不用写哪怕一行代码。 虽然猴子补丁仍然是邪恶的(evil)，但在这种情况下它是“有用的邪恶(useful evil)”。

数据结构

  • 事件
  • 队列
  • 组和池
  • 锁和信号量
  • 线程局部变量
  • 子进程
  • Actors

    事件

事件(event)是一个在Greenlet之间异步通信的形式。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
import gevent
from gevent.event import Event
	
evt = Event()
	
def setter():
    print('A: Hey wait for me, I have to do something')
    gevent.sleep(3)
    print("Ok, I'm done")
    evt.set()
def waiter():
    print("I'll wait for you")
    evt.wait()  # blocking
    print("It's about time")
def main():
    gevent.joinall([
        gevent.spawn(setter),
        gevent.spawn(waiter),
        gevent.spawn(waiter),
        gevent.spawn(waiter)
    ])
if __name__ == '__main__': 
    main()

执行结果：

1
2
3
4
5
6
7
8
A: Hey wait for me, I have to do something
I'll wait for you
I'll wait for you
I'll wait for you
Ok, I'm done
It's about time
It's about time
It's about time

事件对象的一个扩展是AsyncResult，它允许你在唤醒调用上附加一个值。 它有时也被称作是future或defered，因为它持有一个指向将来任意时间可设置为任何值的引用。

1
2
3
4
5
6
7
8
9
10
11
12
13

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),
])

队列

队列是一个排序的数据集合，它有常见的put / get操作， 但是它是以在Greenlet之间可以安全操作的方式来实现的。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
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 xrange(1,10):
        tasks.put_nowait(i)
gevent.spawn(boss).join()
gevent.joinall([
    gevent.spawn(worker, 'steve'),
    gevent.spawn(worker, 'john'),
    gevent.spawn(worker, 'nancy'),
])

执行结果：

1
2
3
4
5
6
7
8
9
10
11
12
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
Quitting time!
Quitting time!
Quitting time!

put和get操作都是阻塞的，put_nowait和get_nowait不会阻塞， 然而在操作不能完成时抛出gevent.queue.Empty或gevent.queue.Full异常。

组和池

组(group)是一个运行中greenlet集合，集合中的greenlet像一个组一样会被共同管理和调度。 它也兼饰了像Python的multiprocessing库那样的平行调度器的角色，主要用在在管理异步任务的时候进行分组。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
import gevent
from gevent.pool import Group
def talk(msg):
    for i in xrange(2):
        print(msg)
g1 = gevent.spawn(talk, 'bar')
g2 = gevent.spawn(talk, 'foo')
g3 = gevent.spawn(talk, 'fizz')
group = Group()
group.add(g1)
group.add(g2)
group.join()
group.add(g3)
group.join()

执行结果：

1
2
3
4
5
6
bar
bar
foo
foo
fizz
fizz

池(pool)是一个为处理数量变化并且需要限制并发的greenlet而设计的结构。

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

执行结果

1
2
3
Size of pool 2
Size of pool 2
Size of pool 1

构造一个socket池的类，在各个socket上轮询。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from gevent.pool import Pool
class SocketPool(object):
    def __init__(self):
        self.pool = Pool(10)
        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:
            self.pool.spawn(self.listen, socket)
    def shutdown(self):
        self.pool.kill()

锁和信号量

信号量是一个允许greenlet相互合作，限制并发访问或运行的低层次的同步原语。 信号量有两个方法，acquire和release。在信号量是否已经被 acquire或release，和拥有资源的数量之间不同，被称为此信号量的范围 (the bound of the semaphore)。如果一个信号量的范围已经降低到0，它会 阻塞acquire操作直到另一个已经获得信号量的greenlet作出释放。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
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:
        print('Worker %i acquired semaphore' % n)
        sleep(0)
    print('Worker %i released semaphore' % n)
pool = Pool()
pool.map(worker1, range(0,2))
`

执行结果：

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

锁(lock)是范围为1的信号量。它向单个greenlet提供了互斥访问。 信号量和锁常被用来保证资源只在程序上下文被单次使用。

线程局部变量

Gevent允许程序员指定局部于greenlet上下文的数据。 在内部，它被实现为以greenlet的getcurrent()为键， 在一个私有命名空间寻址的全局查找。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
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])

执行结果：

1
2
3
1
2
x is not local to f2

很多集成了gevent的web框架将HTTP会话对象以线程局部变量的方式存储在gevent内。 例如使用Werkzeug实用库和它的proxy对象，我们可以创建Flask风格的请求对象。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

from gevent.local import local
from werkzeug.local import LocalProxy
from werkzeug.wrappers import Request
from contextlib import contextmanager
from gevent.wsgi import WSGIServer
_requests = local()
request = LocalProxy(lambda: _requests.request)
@contextmanager
def sessionmanager(environ):
    _requests.request = Request(environ)
    yield
    _requests.request = None
def logic():
    return "Hello " + request.remote_addr
def application(environ, start_response):
    status = '200 OK'
    with sessionmanager(environ):
        body = logic()
    headers = [
        ('Content-Type', 'text/html')
    ]
    start_response(status, headers)
    return [body]
	WSGIServer(('', 8000), application).serve_forever()

子进程

从gevent 1.0起，支持gevent.subprocess，支持协作式的等待子进程。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

很多人也想将gevent和multiprocessing一起使用。最明显的挑战之一 就是multiprocessing提供的进程间通信默认不是协作式的。由于基于 multiprocessing.Connection的对象(例如Pipe)暴露了它们下面的 文件描述符(file descriptor)，gevent.socket.wait_read和wait_write 可以用来在直接读写之前协作式的等待ready-to-read/ready-to-write事件。

```python
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(5):
        msg = b.recv()
        c.send(msg + " in " + str(i))
def put_msg():
    for i in xrange(5):
        wait_write(a.fileno())
        a.send('hi')
def get_msg():
    for i in xrange(5):
        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)
```	  


执行结果：

```shell	
hi in 0
hi in 1
hi in 2
hi in 3
hi in 4

然而要注意，组合multiprocessing和gevent必定带来 依赖于操作系统(os-dependent)的缺陷，其中有：

在兼容POSIX的系统创建子进程(forking)之后， 在子进程的gevent的状态是不适定的(ill-posed)。一个副作用就是， multiprocessing.Process创建之前的greenlet创建动作，会在父进程和子进程两方都运行。

上例的put_msg()中的a.send()可能依然非协作式地阻塞调用的线程：一个 ready-to-write事件只保证写了一个byte。在尝试写完成之前底下的buffer可能是满的。

上面表示的基于wait_write()/wait_read()的方法在Windows上不工作 (IOError: 3 is not a socket (files are not supported))，因为Windows不能监视 pipe事件。

Python包gipc以大体上透明的方式在 兼容POSIX系统和Windows上克服了这些挑战。它提供了gevent感知的基于 multiprocessing.Process的子进程和gevent基于pipe的协作式进程间通信。

Actors

actor模型是一个由于Erlang变得普及的更高层的并发模型。 简单的说它的主要思想就是许多个独立的Actor，每个Actor有一个可以从 其它Actor接收消息的收件箱。Actor内部的主循环遍历它收到的消息，并根据它期望的行为来采取行动。

Gevent没有原生的Actor类型，但在一个子类化的Greenlet内使用队列， 我们可以定义一个非常简单的。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
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)

下面是一个使用的例子：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

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 ZeroMQ
  • 简单server
  • WSGI Servers
  • 流式server
  • Long Polling
  • Websockets

简单server

1
2
3
4
5
6
7
8
9
10
# 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 And Websockets

Gevent为HTTP内容服务提供了两种WSGI server。从今以后就称为 wsgi和pywsgi：

  • gevent.wsgi.WSGIServer
  • gevent.pywsgi.WSGIServer
    glb中使用
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
import click
from flask import Flask
from gevent.pywsgi import WSGIServer
from geventwebsocket.handler import WebSocketHandler
import v1
from .settings import Config
from .sockethandler import handle_websocket
def create_app(config=None):
    app = Flask(__name__, static_folder='static')
    if config:
        app.config.update(config)
    else:
        app.config.from_object(Config)
    app.register_blueprint(
        v1.bp,
        url_prefix='/v1')
    return app
def wsgi_app(environ, start_response):
    path = environ['PATH_INFO']
    if path == '/websocket':
        handle_websocket(environ['wsgi.websocket'])
    else:
        return create_app()(environ, start_response)
@click.command()
@click.option('-h', '--host_port', type=(unicode, int),
    default=('0.0.0.0', 5000), help='Host and port of server.')
@click.option('-r', '--redis', type=(unicode, int, int),
    default=('127.0.0.1', 6379, 0),
    help='Redis url of server.')
@click.option('-p', '--port_range', type=(int, int),
    default=(50000, 61000),
    help='Port range to be assigned.')
def manage(host_port, redis=None, port_range=None):
    Config.REDIS_URL = 'redis://%s:%s/%s' % redis
    Config.PORT_RANGE = port_range
    http_server = WSGIServer(host_port,
                             wsgi_app, handler_class=WebSocketHandler)
    print('----GLB Server run at %s:%s-----' % host_port)
    print('----Redis Server run at %s:%s:%s-----' % redis)
    http_server.serve_forever()

缺陷

和其他异步I/O框架一样,gevent也有一些缺陷:

阻塞(真正的阻塞,在内核级别)在程序中的某个地方停止了所有的东西.这很像C代码中monkey patch没有生效
保持CPU处于繁忙状态.greenlet不是抢占式的,这可能导致其他greenlet不会被调度.
在greenlet之间存在死锁的可能.
一个gevent回避的缺陷是,你几乎不会碰到一个和异步无关的Python库–它将阻塞你的应用程序,因为纯Python库使用的是monkey patch的stdlib.

展开全文 >>

PyZMQ请求应答模式-观察者模式

前面讲过zmq的device,用来充当客户端与服务端的中间件,以增加灵活性，让服务端也变成可插拔。然而device是zmq封装好的，怎样才能一窥内部的数据流呢？看下图
请求应答模式示意图

一看这图就明白了，MonitoredQueue在创建Queue同时，还提供第3个PUB socket来发布途经这个Queue的进出信息。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
import time
import zmq
from zmq.devices.basedevice import ProcessDevice
from zmq.devices.monitoredqueuedevice import MonitoredQueue
from zmq.utils.strtypes import asbytes
from multiprocessing import Process
import random
  
frontend_port = 5559  
backend_port = 5560  
monitor_port = 5562  
number_of_workers = 2

def monitordevice():  
    in_prefix=asbytes('in')  
    out_prefix=asbytes('out')  
    monitoringdevice = MonitoredQueue(zmq.XREP, zmq.XREQ, zmq.PUB, in_prefix, out_prefix)  
      
    monitoringdevice.bind_in("tcp://127.0.0.1:%d" % frontend_port)  
    monitoringdevice.bind_out("tcp://127.0.0.1:%d" % backend_port)  
    monitoringdevice.bind_mon("tcp://127.0.0.1:%d" % monitor_port)  
      
    monitoringdevice.setsockopt_in(zmq.RCVHWM, 1)  #对进入socket的消息设置高水位
    monitoringdevice.setsockopt_out(zmq.SNDHWM, 1) #对向外发送的消息设置高水位（最大缓存量）
    monitoringdevice.start()    
    print("Program: Monitoring device has started")

def server(backend_port):  
    print("Program: Server connecting to device") 
    context = zmq.Context()  
    socket = context.socket(zmq.REP)  
    socket.connect("tcp://127.0.0.1:%s" % backend_port)  
    server_id = random.randrange(1,10005)  
    while True:  
        message = socket.recv()  
        print("Server: Received - %s" % message)  
        socket.send_string("Response from server #%s" % server_id)
def client(frontend_port, client_id):  
    print("Program: Worker #%s connecting to device" % client_id)
    context = zmq.Context()  
    socket = context.socket(zmq.REQ)  
    socket.connect("tcp://127.0.0.1:%s" % frontend_port)  
    request_num = 1  
    socket.send_string("Request #%s from client#%s" % (request_num, client_id))  
    #  Get the reply.  
    message = socket.recv_multipart()  
    print("Client: Received - %s" % message)

def monitor():  
    print("Starting monitoring process") 
    context = zmq.Context()  
    socket = context.socket(zmq.SUB)  
    print( "Collecting updates from server...")  
    socket.connect("tcp://127.0.0.1:%s" % monitor_port)  
    socket.setsockopt(zmq.SUBSCRIBE, b"")  
    while True:  
        string = socket.recv_multipart()  
        print("Monitoring Client: %s" % string)
if __name__ == '__main__':
    monitoring_p = Process(target=monitordevice)  
    monitoring_p.start()    
    server_p = Process(target=server, args=(backend_port,))  
    server_p.start()    
    monitorclient_p = Process(target=monitor)  
    monitorclient_p.start()    
    time.sleep(2)     
  
    for client_id in range(number_of_workers):  
        Process(target=client, args=(frontend_port, client_id,)).start()

    time.sleep(10)
    server_p.terminate()
    monitorclient_p.terminate()
    monitoring_p.terminate()

展开全文 >>

PyZMQ一种可靠的请求应答模式--超时重试模式

提升REQ/REP模式的客户端侧可靠性，由于在此模式下，客户端与服务端严格遵循你一下我一下的乒乓规则，当然现实中不可能这么配合，比如服务器突然挂了，客户端就会阻塞在socket.recv(),然后过段时间，服务器又ok了，客户端也自动重连。但是有的场景客户端需要及时知道服务端状况而不是在那一直等，需要返回个错误之类。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
### 请求端
import sys  
import zmq  
  
REQUEST_TIMEOUT = 2500  
REQUEST_RETRIES = 3  
SERVER_ENDPOINT = "tcp://localhost:5555"  
  
context = zmq.Context(1)  
  
prin("I: Connecting to server…") 
client = context.socket(zmq.REQ)  
client.connect(SERVER_ENDPOINT)  
  
poll = zmq.Poller()  
poll.register(client, zmq.POLLIN)  
  
sequence = 0  
retries_left = REQUEST_RETRIES  
while retries_left:  
    sequence += 1  
    request = str(sequence)  
    print("I: Sending (%s)" % request)
    client.send(request)  
  
    expect_reply = True  
    while expect_reply:  
        socks = dict(poll.poll(REQUEST_TIMEOUT))  
        if socks.get(client) == zmq.POLLIN:  
            reply = client.recv()  
            if not reply:  
                break  
            if int(reply) == sequence:  
                prin("I: Server replied OK (%s)" % reply) 
                retries_left = REQUEST_RETRIES  
                expect_reply = False  
            else:  
                prin("E: Malformed reply from server: %s" % reply)
  
        else:  
            print("W: No response from server, retrying…")  
            # Socket is confused. Close and remove it.  
            client.setsockopt(zmq.LINGER, 0)  
            client.close()  
            poll.unregister(client)  
            retries_left -= 1  
            if retries_left == 0:  
                print("E: Server seems to be offline, abandoning") 
                break  
            print("I: Reconnecting and resending (%s)" % request) 
            # Create new connection  
            client = context.socket(zmq.REQ)  
            client.connect(SERVER_ENDPOINT)  
            poll.register(client, zmq.POLLIN)  
            client.send(request)  
  
context.term()

重点是：使用poll.poll(REQUEST_TIMEOUT)，来执行一个限时读操作,如果时间段内该REQ socket没收到数据，视作服务端故障,这时就尝试先关闭当前socket,然后重连服务端,重发request。

另外,注意client.setsockopt(zmq.LINGER, 0)设置linger为0，即马上丢掉socket中未处理的消息,否则context,term()啥时能执行完就不知道了。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
# 服务于端
#coding=utf-8  
  
from random import randint  
import time  
import zmq  
  
context = zmq.Context(1)  
server = context.socket(zmq.REP)  
server.bind("tcp://*:5555")  
  
cycles = 0  
while True:  
    request = server.recv()  
    cycles += 1  
    # Simulate various problems, after a few cycles  
    if cycles > 3 and randint(0, 3) == 0:  
        print("I: Simulating a crash") 
        break  
    elif cycles > 3 and randint(0, 3) == 0:  
        print("I: Simulating CPU overload")
        time.sleep(2)  
        break  
    print("I: Normal request (%s)" % request) 
    time.sleep(1) # Do some heavy work  
    server.send(request)  
  
server.close()  
context.term()

展开全文 >>

ZeroMQ简介

概述

引用官方的说法： “ZMQ (以下 ZeroMQ 简称 ZMQ)是一个简单好用的传输层，像框架一样的一个 socket library，他使得 Socket 编程更加简单、简洁和性能更高。是一个消息处理队列库，可在多个线程、内核和主机盒之间弹性伸缩。ZMQ 的明确目标是“成为标准网络协议栈的一部分，之后进入 Linux 内核”。现在还未看到它们的成功。但是，它无疑是极具前景的、并且是人们更加需要的“传统”BSD 套接字之上的一层封装。ZMQ 让编写高性能网络应用程序极为简单和有趣。”

近几年有关”Message Queue”的项目层出不穷，知名的就有十几种，这主要是因为后摩尔定律时代，分布式处理逐渐成为主流，业界需要一套标准来解决分布式计算环境中节点之间的消息通信。几年的竞争下来，Apache 基金会旗下的符合 AMQP/1.0标准的 RabbitMQ 已经得到了广泛的认可，成为领先的 MQ 项目。

与 RabbitMQ 相比，ZMQ 并不像是一个传统意义上的消息队列服务器，事实上，它也根本不是一个服务器，它更像是一个底层的网络通讯库，在 Socket API 之上做了一层封装，将网络通讯、进程通讯和线程通讯抽象为统一的 API 接口。

消息模型

ZMQ 提供了三个基本的通信模型，分别是Request-Reply，Publisher-Subscriber，Parallel Pipeline,我们从这三种模式一窥 ZMQ 的究竟

请求应答模式

由 Client 发起请求，并等待 Server 回应请求。请求端发送一个简单的 hello，服务端则回应一个 world。请求端和服务端都可以是 1:N 的模型。通常把 1 认为是 Server ，N 认为是 Client 。ZMQ 可以很好的支持路由功能（实现路由功能的组件叫作 Device），把 1:N 扩展为N:M （只需要加入若干路由节点。如下图

请求应答模式示意图

1
2
3
4
5
6
7
8
9
10
11
12
13
#响应端代码
# -*- coding:utf-8 -*-
import zmq
import time

context = zmq.Context()
socket = context.socket(zmq.REP)
socket.bind("tcp://*:5555")
while True:
    message = socket.recv()
    print("Received request:%s", %(message))
    time.sleep (1)       
    socket.send(time_utils.get_format_timestamp()+" World")
1
2
3
4
5
6
7
8
9
10
#请求端代码
# -*- coding:utf-8 -*-
import zmq
context = zmq.Context()
socket = context.socket(zmq.REQ)
socket.connect ("tcp://localhost:5555")
for request in range (1,10):
    socket.send ("Hello")
    message = socket.recv()
    print("Received reply ：%s"%(message))
  1. 服务端和客户端无论谁先启动，效果是相同的，这点不同于 Socket。
  2. 在服务端收到信息以前，程序是阻塞的，会一直等待客户端连接上来。
  3. 服务端收到信息以后，会 send 一个“World”给客户端。值得注意的是一定是 client 连接上来以后，send 消息给 Server，然后 Server 再 rev 然后响应 client，这种一问一答式的。如果 Server 先 send，client 先 rev 是会报错的。
  4. ZMQ 通信通信单元是消息，他除了知道 Bytes 的大小，他并不关心的消息格式。因此，你可以使用任何你觉得好用的数据格式。Xml、Protocol Buffers、Thrift、json 等等。
    虽然可以使用 ZMQ 实现 HTTP 协议，但是，这绝不是他所擅长的。
  5. 通常bind为绑定本地的端口，而connect为连接远程端口，不一定是服务端绑定，客户端连接，至于谁绑定谁连接要根据实际情况来决定。
  6. 请求发出必须有应答不然会报错，相应端尚未完成一个请求响应不能接受第二请求

发布订阅模型

我们可以想象一下天气预报的订阅模式，由一个节点提供信息源，由其他的节点，接受信息源的信息

请求应答模式示意图

1
2
3
4
5
6
7
8
9
10
11
12
13
#发布端代码
# -*- coding:utf-8 -*-
import zmq
import time
from common import time_utils
context = zmq.Context()
socket = context.socket(zmq.PUB)
socket.bind("tcp://*:5555")
while True:
    message = time_utils.get_format_timestamp()+":Publish message"
    print("Publish Message:%s ", %(message))
    time.sleep (1)
    socket.send(message)
1
2
3
4
5
6
7
8
9
10
# -*- coding:utf-8 -*-

import zmq
context = zmq.Context()
socket = context.socket(zmq.SUB)
socket.connect ("tcp://localhost:5555")
socket.setsockopt(zmq.SUBSCRIBE, '')
while True:
    message = socket.recv()
    print("Receive Message:%s"+%(message))
  1. 与请求应答模式不同的是Socket 的类型变成 SOCKET_PUB 和 SOCKET_SUB 类型。客户端需要subscriber−>setSockOpt(ZMQ::SOCKOPTSUBSCRIBE,filter);设置一个过滤值，相当于设定一个订阅频道，否则什么信息也收不到。
  2. 服务器端一直不断的广播中，如果中途有 Subscriber 端退出，并不影响他继续的广播，当 Subscriber 再连接上来的时候，收到的就是后来发送的新的信息了。这对比较晚加入的，或者是中途离开的订阅者，必然会丢失掉一部分信息。
    但是，如果 Publisher 中途离开，所有的 Subscriber 会阻塞住，等待 Publisher 再上线的时候，会继续接受信息。

管道模型

这个是 PUSH/PULL 模式，又叫做pipeline管道模式,取其一去不回头之意，一推一拉，数据滚滚向前。
这种socket封装的原本用意呢，是把数据交给一组worker端干活，PUSH会把任务均匀的（这个好像是zmq的招牌）的分配给下游的worker们，保证大家都有活干，如下图：
请求应答模式示意图

Producer:产生代处理数据，并将数据push出去

consumer(worker):pull到来自producer的数据，处理后,push出去

resultcolltor:pull到consumer的数据处理结果

1
2
3
4
5
6
7
8
9
10
11
12
13
# 生产者

import time  
import zmq  
  
def producer():  
    context = zmq.Context()  
    zmq_socket = context.socket(zmq.PUSH)  
    zmq_socket.bind("tcp://127.0.0.1:5557")  
    for num in xrange(2000):  
        work_message = { 'num' : num }  
        zmq_socket.send_json(work_message)  
producer()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
# 消费者

import time  
import zmq  
import random  
  
def consumer():  
    consumer_id = random.randrange(1,10005)  
    print "I am consumer #%s" % (consumer_id)  
    context = zmq.Context()  
    # recieve work  
    consumer_receiver = context.socket(zmq.PULL)  
    consumer_receiver.connect("tcp://127.0.0.1:5557")  
    # send work  
    consumer_sender = context.socket(zmq.PUSH)  
    consumer_sender.connect("tcp://127.0.0.1:5558")  
      
    while True:  
        work = consumer_receiver.recv_json()  
        data = work['num']  
        result = { 'consumer' : consumer_id, 'num' : data}  
        if data%2 == 0:   
            consumer_sender.send_json(result)  
consumer()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
#汇总收集者
import time  
import zmq  
import pprint  
  
def result_collector():  
    context = zmq.Context()  
    results_receiver = context.socket(zmq.PULL)  
    results_receiver.bind("tcp://127.0.0.1:5558")  
    collecter_data = {}  
    for x in xrange(1000):  
        result = results_receiver.recv_json()  
        if collecter_data.has_key(result['consumer']):  
            collecter_data[result['consumer']] = collecter_data[result['consumer']] + 1  
        else:  
            collecter_data[result['consumer']] = 1  
        if x == 999:  
            pprint.pprint(collecter_data)  
  
result_collector()

生产者 使用的是 SOCKET_PUSH，将任务分给消费者节点上。而消费者使用 SOCKET_PULL 从上游接受任务，并使用 SOCKET_PUSH 将结果汇集到 收集者。值得注意的是，任务的分发的时候也同样有一个负载均衡的路由功能，消费者可以随时自由加入，生产者可以均衡将任务分发出去。

与发布订阅不同的是，生产者发布出去消息如果没有被即使消费，那么会直接保存在缓存中，直到消费者上线将消息消费了。但是如果长期没有消费者消费消息，当达到一定的标准是，生产者会将最开始的数据丢弃，整个消息的消费遵循先进先出原则

展开全文 >>

一种基于django-restFramework通用型增删改查方案

方案理论基础

在Web开发中对于数据模型的操作不可避免，常见的数据资源的对象一般有两种：

  1. 对某一数据资源集合的操作
  2. 对某一数据资源对象的的操作

对于数据资源操作及其前端HTTP请求方法通常有：

  1. 增—-> POST or PUT
  2. 删—-> DELETE
  3. 改—-> POST or PUT
  4. 查—-> POST or GET

而业界RESTful API设计风格的对资源的操作类型和HTTP动词关系设计通常采用如下方案：

  1. 增—-> POST (在服务器新建资源)
  2. 删—-> DELETE (从服务器删除资源)
  3. 改—-> PUT (在服务器更新资源，客户端提供资源改变属性)
  4. 查—-> GET (从服务器获去资源)
  5. 改—-> PATCH (在服务器更新资源，服务器提供资源改变属性)

利用上述原理和django-restFramework的序列化器serialiaer我们设计一种适用于django框架项目的通用型增删改查方案。

项目架构

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
project---
         |--project--|
                     |__init__.py
                     |urls.py
                     |settings.py
                     |wsgi.py

         |---app-----|__init__.py
                     |app.py
                     |urls.py
                     |views.py
                     |serializers.py     
                     |models.py
                     |admin.py
        ....

        |--manage.py

实现方案

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
#views.py
from django.shortcuts import render
from django.http import HttpResponse
from rest_framework import status
from django.db.models import Q
from functools import reduce
import operator
from app import serializers
from app import model
from app.serializers import JSONResponse
from rest_framework.parsers import JSONParserl
import importlib

def model_list(request):
     
    """
    列出所有的code snippet，或创建一个新的snippet。
    """
    model_str=request.META.get('HTTP_MODEL')
    global model
    global serializers
    if model_str:
        Model=getattr(model,model_str)
        ModelSerializer=getattr(serializers,model_str+'Serializer')
        if request.method == 'GET':
            # 获取资源列表
            models = Model.objects.ALL()
            serializer = ModelSerializer(models, many=True)
            return JSONResponse(serializer.data)

        elif request.method == 'POST':
            # 根据条件获取资源列表
            data = JSONParser().parse(request))
            serializer = ModelSerializer(data=data)
            fields=Model.get_fields()
            q_list=[]
            for field in fields:
                value=data.get(field)
                if value:
                    q_list.append(Q(**{field+'__icontains':value}))
            id_list=Model.get_indexs()
            for field in id_list:
                value=data.get(field)
                if isinstance(value,list):
                    if data.get(field):
                        q_list.append(Q(**{field+'__in':value}))
                if isinstance(value,int):
                    if data.get(field):
                        q_list.append(Q(**{field+'':value}))
                if isinstance(value,str):
                    if data.get(field):
                        q_list.append(Q(**{field+'':int(value)}))
            models=Model.objects.filter((reduce(operator.and_, q_list)))
            serializer = ModelSerializer(models, many=True)
            return JSONResponse(serializer.data)
        elif request.method == 'PUT':
        # 新建资源
            data = JSONParser().parse(request)
            serializer = ModelSerializer(data=data)
            if serializer.is_valid():
                serializer.save()
                return JSONResponse(serializer.data, status=status.HTTP_201_CREATED)
            return JSONResponse(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
    return JSONResponse('not definde request headers model', status=status.HTTP_400_BAD_REQUEST)



def model_detail(request,pk):
    """
    获取，更新或删除一个 code snippet。
    """
    model_str=request.META.get('HTTP_MODEL')
    global model
    global serializers
    if model_str:
        Model=getattr(model,model_str)
        ModelSerializer=getattr(serializers,model_str+'Serializer')
        try:
            model = Model.objects.get(pk=pk)
        except Model.DoesNotExist:
            return HttpResponse(status=status.HTTP_404_NOT_FOUND)

        if request.method == 'GET':
        #获取一个资源对象
            serializer = ModelSerializer(model)
            return JSONResponse(serializer.data)

        elif request.method == 'PUT':
        # 更新一个资源对象
            data = JSONParser().parse(request)
            serializer = ModelSerializer(model, data=data)
            if serializer.is_valid():
                serializer.save()
                return Response(serializer.data)
            return JSONResponse(serializer.errors, status=status.HTTP_400_BAD_REQUEST)

        elif request.method == 'DELETE':
        #删除一个资源对象
            model.delete()
            return HttpResponse(status=status.HTTP_204_NO_CONTENT)
    return JSONResponse('not definde request headers model', status=status.HTTP_400_BAD_REQUEST)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
# serializers.py
from rest_framework import serializers
from rest_framework.renderers import JSONRenderer
from app.model import Modal
from django.http import HttpResponse

class JSONResponse(HttpResponse):
    """
    An HttpResponse that renders its content into JSON.
    """
    def __init__(self, data, **kwargs):
        content = JSONRenderer().render(data)
        kwargs['content_type'] = 'application/json'
        super(JSONResponse, self).__init__(content, **kwargs)

class ModalSerializer(serializers.ModelSerializer):
    class Meta:
        model = Modal
        fields = '__all__'

请求封装

首先在前端请求的headers中添加我们需要获取的资源类名

1
2
3
4
5
import requests
headers={'model':'Model','Content-Type':'application/json'}
url='http://127.0.0.1:8000/app/model/model_list/'
response=requests.get(url, headers=headers)
print(response.text)

为了防止django拦截headers中的参数，我们需要在settings文件中配置request跨域的允许访问的headers字段

1
2
3
4
5
CORS_ALLOW_HEADERS = (
    'accept',
    ...,
    'model',
)

解析并获取资源类

前台请求制定资源对象类名和请求HTTP动词即可通过反射定位到对应方法，完成对资源的相关操作

1
2
3
4
5
6
7
# app 下的model文件
from app import model
from app import serializers
# 反射获取model 类
Model=getattr(model,model_str)
# 反射获取 model类对象的序列化对象
ModelSerializer=getattr(serializers,model_str+'Serializer')

动态构建资源对象的查询条件

通常我们对资源查询对于像id、pk、fk等索引性条件采用field__in=list_obj或者field=int_obj,对于其他字段通常采用field__icontains=obj的模糊匹配的方法，进行条件查询，利用这些特性，我们使用Q查询动态构建查询条件。

为了让改方法适用于所有的模型，我们讲查询条件定义到 ｀model.py｀中的资源对象类中静态方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
class UserDetail(models.Model):
    name=models.CharField(max_length=50,  null=True)
    sex=models.CharField(max_length=2, null=True)
    age=models.IntegerField()
    address=models.CharField(max_length=100, null=True)
    user=models.ForeignKey(User,null=True, blank=True, on_delete=models.SET_NULL))

    @staticmethod
    def get_indexs():
        return ['id''age','pk']

    @staticmethod
    def get_fields(Self):
        return ['name','sex','address']

至此我们基本完成了jango项目通用型增删改查的方案，所有跟资源模型相关在model文件中定义，如果要使用django-restframwork序列化方法的相关则在serializers.py中定义，对资源模型的操作在views文件中定义，当然你完全可以重新定义当多个APP使用同一个model_list()和model_detail()方法，也可以不采用django-restframwork

展开全文 >>

Django动态路由系统实现方案

在django开发过程中，比较规范的路由通常遵循这样的规则

1
app-->model-->method 即('app\model\method')

采用反射机制可以为项目设置动态路由，避免了大量的URL文件和URL维护，缩短了开发周期

django项目的架构通常是这样

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
project---
         |--project--|
                     |__init__.py
                     |urls.py
                     |settings.py
                     |wsgi.py

         |---app-----|__init__.py
                     |app.py
                     |urls.py
                     |views--|__init__.py
                             |model1.py
                             |model2.py
                             |...
                             
                     |model.py
                     |admin.py
        ....

        |--manage.py

在project文件夹下建立activater.py路由匹配文件,内容如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
#!/usr/bin/env python
#coding:utf-8
import importlib
from django.shortcuts import render_to_response,HttpResponse,redirect


def process(request,**kwargs):
    '''接收所有匹配url的请求，根据请求url中的参数，通过反射动态指定view中的方法'''
    
    app =  kwargs.get('app',None)
    function = kwargs.get('function',None)
    model = kwargs.get('model',None)
    try:
        app_obj=__import__("%s.views" %app)
        view_obj = getattr(app_obj, 'views')
        model_obj = getattr(view_obj, model)
        func_obj = getattr(model_obj, function)
    #执行view.py中的函数，并获取其返回值
        if kwargs.get('id'):
            result=func_obj(request,kwargs.get('id'))
        else:
            result=func_obj(request)
    except (ImportError,AttributeError) as e:
        print(e)
        #导入失败时，自定义404错误
        return HttpResponse('404 Not Found')
    except Exception as e:
        print(e)
        #代码执行异常时，自动跳转到指定页面
        return HttpResponse('404 Not Found')
    
    return result

修改project文件夹下的urls.py为：

1
2
3
4
5
6
7
8
9
10
from django.contrib import admin
from django.urls import path,include,re_path
from .activater import process

urlpatterns = [
    path('admin/', admin.site.urls),
    re_path('^(?P<app>(\w+))/(?P<model>(\w+))/(?P<function>(\w+))/(?P<id>(\d+))/$',process),
    re_path('^(?P<app>(\w+))/(?P<model>(\w+))/(?P<function>(\w+))/$',process),
]
`

修改后的项目架构为：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
project---
         |--project--|
                     |__init__.py
                     |urls.py
                     |settings.py
                     |wsgi.py
                     |activater.py

         |---app-----|__init__.py
                     |app.py
                     |views--|__init__.py
                             |model1.py
                             |model2.py
                             |...

                     |model.py
                     |admin.py
        ....

        |--manage.py

至此django一个动态路由系统就实现了，当然实际在开发过程中需要根据项目架构修改activater.py映射关系，不必要按照本文代码来

note:本代码示例基于django2.0版本

展开全文 >>

Express入门

什么是Express？

基于 Node.js 平台，快速、开放、极简的 Web 开发框架
Express 是一个保持最小规模的灵活的 Node.js Web 应用程序开发框架，为 Web 和移动应用程序提供一组强大的功能。

安装

1
npm init

此命令将要求你输入几个参数，例如此应用的名称和版本。 你可以直接按“回车”键接受大部分默认设置即可，下面这个除外：

1
entry point: (index.js)

键入 app.js 或者你所希望的名称，这是当前应用的入口文件。如果你希望采用默认的 index.js 文件名，只需按“回车”键即可。
会生成一个package.json文件，具体如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
{
  "name": "demo1",
  "version": "1.0.0",
  "description": "",
  "main": "app.js",
  "scripts": {
    "test": "echo \"Error: no test specified\" && exit 1"
  },
  "author": "",
  "license": "ISC",
  "dependencies": {
    "express": "^4.17.0"
  }
}

从这个文件我们可以看出，这个程序名字是demo1，程序的主入口为app.js,接下来安装Express

1
npm install express --save

编写主应用程序app.js

1
2
3
4
const express = require('express')
const app = express()
app.get('/', (req, res) => res.send('Hello World!'))
app.listen(3000, () => console.log('Example app listening on port 3000!'))

访问

打开浏览器，访问127.0.0.1:3000，即可看到Hello World

Express 应用生成器

安装

1
sudo install express-generator -g

输入express -h,可以看到常用命令

生成应用程序

创建一个名为myapp的Express 应用，并使用pug为模板引擎

1
express --view=pug myapp

运行

liux运行下面的命令

1
2
3
cd myapp
npm install
DEBUG=myapp:* npm start

windows运行下面命令

1
2
3
cd myapp
npm install
set DEBUG=myapp:* & npm start

在浏览器打开｀127.0.0.1:3000｀即可看到次应用程序

展开全文 >>

防止axios发起重复请求

前言

在web开发中，我们经常遇到用户重复点击导致某个请求多次提交的问题，今天我们分享两种axios防止用户提交重复请求的方法。

如何认定为重复请求？

当一次请求尚未完成又发起的相同的请求的情况，我们称之为重复请求。

解决思路

  1. 当上次请求尚未完成，用户再次发起请求的时候，我们将上次请求取消，并开始新的请求。
  2. 当上次请求尚未完成，如果用户再次发起相同的请求，我们将本次请求取消，等待上次请求完成。
  3. 两种方式都必须用到了axios路由拦截器，在请求之前进行判别和请求之后进行去除
  4. 其他方式

以下是实现思路图：
axios防止重复请求示意图

more >>

展开全文 >>