改善 Python 程序的 91 个建议（六）

$ python -m timeit -n 1000 "[x for x in range(1000) if x in     range(600, 1000)]"
1000 loops, best of 3: 6.44 msec per loop
$ python -m timeit -n 1000 "set(range(100)).intersection(range(60, 100))"   
1000 loops, best of 3: 9.18 usec per loop

from multiprocessing import Process, Pipe, Queue
import time

def reader_pipe(pipe):
    output_p, input_p = pipe    # 返回管道的两端
    inout_p.close()
    while True:
        try:
            msg = output_p.recv()    # 从 pipe 中读取消息
        except EOFError:
                break

def writer_pipe(count, input_p):    # 写消息到管道中
    for i in range(0, count):
        input_p.send(i)                # 发送消息

def reader_queue(queue):            # 利用队列来发送消息
    while True:
        msg = queue.get()            # 从队列中获取元素
        if msg == "DONE":
            break

def writer_queue(count, queue):
    for ii in range(0, count):
        queue.put(ii)                # 放入消息队列中
    queue.put("DONE")

if __name__ == "__main__":
    print("testing for pipe:")
    for count in [10 ** 3, 10 ** 4, 10 ** 5]:
        output_p, input_p = Pipe()
        reader_p = Process(target=reader_pipe, args=((output_p,     input_p),))
        reader_p.start()            # 启动进程
        output_p.close()
        _start = time.time()
        writer_pipe(count, input_p)    # 写消息到管道中
        input_p.close()
        reader_p.join()                # 等待进程处理完毕
        print("Sending {} numbers to Pipe() took {}     seconds".format(count, (time.time() - _start)))

    print("testsing for queue:")
    for count in [10 ** 3, 10 ** 4, 10 ** 5]:
        queue = Queue()                # 利用 queue 进行通信
        reader_p = Process(target=reader_queue, args=((queue),))
        reader_p.daemon = True
        reader_p.start()

        _start = time.time()
        writer_queue(count, queue)    # 写消息到 queue 中
        reader_p.join()
        print("Seding {} numbers to Queue() took {}     seconds".format(count, (time.time() - _start)))

# 示例一
import time
from multiprocessing import Process, Value

def func(val):    # 多个进程同时修改 val
    for i in range(10):
        time.sleep(0.1)
        val.value += 1

if __name__ == "__main__":
    v = Value("i", 0)    # 使用 value 来共享内存
    processList = [Process(target=func, args=(v,)) for i in     range(10)]
    for p in processList: p.start()
    for p in processList: p.join()
    print v.value
# 修改 func 函数，真正控制同步访问
def func(val):
    for i in range(10):
        time.sleep(0.1)
        with val.get_lock():    # 仍然需要使用 get_lock 方法来获取锁对象
            val.value += 1
# 示例二
import multiprocessing
def f(ns):
    ns.x.append(1)
    ns.y.append("a")

if __name__ == "__main__":
    manager = multiprocessing.Manager()
    ns = manager.Namespace()
    ns.x = []    # manager 内部包括可变对象
    ns.y = []

    print("before process operation: {}".format(ns))
    p = multiprocessing.Process(target=f, args=(ns,))
    p.start()
    p.join()
    print("after process operation {}".format(ns))    # 修改根本不会生效
# 修改
import multiprocessing
def f(ns, x, y):
    x.append(1)
    y.append("a")
    ns.x = x    # 将可变对象也作为参数传入
    ns.y = y

if __name__ == "__main__":
    manager = multiprocessing.Manager()
    ns = manager.Namespace()
    ns.x = []    # manager 内部包括可变对象
    ns.y = []

    print("before process operation: {}".format(ns))
    p = multiprocessing.Process(target=f, args=(ns, ns.x, ns.y))
    p.start()
    p.join()
    print("after process operation {}".format(ns))

f = None
def child(f):
    # do something

if __name__ == "__main__":
    f = open(filename, mode)
    p = Process(target=child)
    p.start()
    p.join()
# 推荐的方式
def child(f):
    print(f)

if __name__ == "__main__":
    f = open(filename, mode)
    p = Process(target=child, args=(f, ))    # 将资源对象作为构造函数参数传入
    p.start()
    p.join()

import multiprocessing
def unwrap_self_f(*args, **kwargs):
    return calculate.f(*args, **kwargs)    # 返回一个对象

class calculate(object):
    def f(self, x):
        return x * x
    def run(self):
        p = multiprocessing.Pool()
        return p.map(unwrap_self_f, zip([self] * 3, [1, 2, 3]))

if __name__ == "__main__":
    c1 = calculate()
    print(c1.run())

import urllib2
import os
import time
import threadpool

def download_file(url):
    print("begin download {}".format(url ))
    urlhandler = urllib2.urlopen(url)
    fname = os.path.basename(url) + ".html"
    with open(fname, "wb") as f:
        while True：
            chunk = urlhandler.read(1024)
            if not chunk:
                break
            f.write(chunk)

urls = ["http://wiki.python.org/moni/WebProgramming",
    "https://www.createspace.com/3611970",
    "http://wiki.python.org/moin/Documention"]
pool_size = 2
pool = threadpool.ThreadPool(pool_size)    # 创建线程池，大小为 2
requests = threadpool.makrRequests(download_file, urls)    # 创建工作请求
[pool.putRequest(req) for req in requests]

print("putting request to pool")
pool.putRequest(threadpool.WorkRequest(download_file, args=["http://chrisarndt.de/projects/threadpool/api/",]))    # 将具体的请求放入线程池
pool.putRequest(threadpool.WorkRequest(download_file, args=["https://pypi.python.org/pypi/threadpool",]))
pool.poll()    # 处理任务队列中的新的请求
pool.wait()
print("destory all threads before exist")
pool.dismissWorkers(pool_size, do_join=True)    # 完成后退出

#include "Python.h"
static PyObject * pr_isprime(PyObject, *self, PyObject * args) {
int n, num;
if (!PyArg_ParseTuple(args, "i", &num))    // 解析参数
    return NULL;
if (num < 1) {
    return Py_BuildValue("i", 0);    // C 类型的数据结构转换成 Python 对象
}
n = num - 1;
while (n > 1) {
    if (num % n == 0) {
    return Py_BuildValue("i", 0);
    n--;
    }
}
return Py_BuildValue("i", 1);
}

static PyMethodDef PrMethods[] = {
{"isPrime", pr_isprime, METH_VARARGS, "check if an input number is prime or not."},
{NULL, NULL, 0, NULL}
};

void initpr(void) {
(void) Py_InitModule("pr", PrMethods);
}

struct PyMethodDef {
    char * m1_name;        // 方法名
    PyCFunction m1_meth;    // 导出函数
    int m1_flags;            // 参数传递方法
    char * m1_doc;        // 方法描述
} 参数传递方法一般设置为 METH_VARARGS，如果想传入关键字参数，则可以将其与 METH_KEYWORDS 进行或运算。若不想接受任何参数，则可以将其设置为 METH_NOARGS。该结构体必须与 {NULL, NULL, 0, NULL} 所表示的一条空记录来结尾。

from distutils.core import setup, Extension
module = Extension("pr", sources=["testextend.c"])
setup(name="Pr test", version="1.0", ext_modules=[module])

# 安装
$ pip install -U cython
# 生成 .c 文件
$ cython arithmetic.py
# 提交编译器
$ gcc -shared -pthread -fPIC -fwrapv -02 -Wall -fno-strict-aliasing -I /usr/include/python2.7 -o arithmetic.so arithmetic.c
# 这时生成了 arithmetic.so 文件
# 我们就可以像 import 普通模块一样使用它

$ cp arithmetic.py arithmetic.pyx
$ cd ~
$ python
>>> import pyximport; pyximport.install()
>>> import arithmetic
>>> arithmetic.__file__

import math
def great_circle(lon1, lat1, lon2, lat2):
    radius = 3956    # miles
    x = math.pi / 180.0
    a = (90.0 - lat1) * (x)
    b = (90.0 - lat2) * (x)
    theta = (lon2 - lon1) * (x)
    c = math.acos(math.cos(a) * math.cos(b)) + (math.sin(a) *   math.sin(b) * math.cos(theta))
    return radius * c

import math
def great_circle(float lon1, float lat1, float lon2, float lat2):
    cdef float radius = 3956.0
    cdef float pi = 3.14159265
    cdef float x = pi / 180.0
    cdef float a, b, theta, c
    a = (90.0 - lat1) * (x)
    b = (90.0 - lat2) * (x)
    theta = (lon2 - lon1) * (x)
    c = math.acos(math.cos(a) * math.cos(b)) + (math.sin(a) * math.sin(b) * math.cos(theta))
    return radius * c

cdef extern from "math.h":
    float cosf(float theta)
    float sinf(float theta)
    float acosf(float theta)

def greate_circle(float lon1, float lat1, float lon2, float lat2):
    cdef float radius = 3956.0
    cdef float pi = 3.14159265
    cdef float x = pi / 180.0
    cdef float a, b, theta, c
    a = (90.0 - lat1) * (x)
    b = (90.0 - lat2) * (x)
    theta = (lon2 - lon1) * (x)
    c = acosf((cosf(a) * cosf(b)) + (sinf(a) * sinf(b) * cosf(theta)))
    return radius * c