多进程，多线程，协程

多进程，多线程，协程 多进程 linux系统可通过os.fork()复制当前进程状态作为子进程。复制时子进程返回0,父进程返回子进程的pid. 子进程可通过os.getppid()获取父进程的pid.同时os.getpid()可获得当前进程的pid.

    import os

    print 'Process (%s) start...' % os.getpid()
    pid = os.fork()
    if pid==0:
        print 'I am child process (%s) and my parent is %s.' % (os.getpid(),                os.getppid())
    else:
    print 'I (%s) just created a child process (%s).' % (os.getpid(), pid)

结果:

Process (876) start...
I (876) just created a child process (877).
I am child process (877) and my parent is 876.

windows没有fork().可以通过python提供的通用多进程模块multiprocessing创建多进程.

创建多进程需要导入Process模块:

from multiprocess import Process

使用

p = Process(target=function, args=(parament,...)

创建子进程实例.其中target=传入子进程需执行的函数本身function,args传入函数需要的参数.参数数量不固定.
之后使用

p.start()

运行实例.要等待该子进程运行结束再运行之后的代码可以使用:

p.join()

以下是一个例子:

from multiprocessing import Process
import os


# 子进程要执行的代码

def run_proc(name):
    print 'Run child process %s (%s)...' % (name, os.getpid())

if __name__=='__main__':
    print 'Parent process %s.' % os.getpid()
    p = Process(target=run_proc, args=('test',))
    print 'Process will start.'
    p.start()
    p.join()
    print 'Process end.'

结果:

Parent process 928.
Process will start.
Run child process test (929)...
Process end.

对于需启动大量子进程的情况,可使用Pool模块:

from multiprocessing import Pool

使用:

p = Pool(number)

创建进程池.其中number为进程池包含子进程数量.不写默认为CPU核数.

使用:

p.apply_async(function, args=(parament,...)

运行子进程.

之后需关闭进程池:

p.close()

同时,需等待所有子进程运行结束可使用:

p.join()

以下是一个例子:

from multiprocessing import Pool
import os, time, random

def long_time_task(name):
    print 'Run task %s (%s)...' % (name, os.getpid())
    start = time.time()
    time.sleep(random.random() * 3)
    end = time.time()
    print 'Task %s runs %0.2f seconds.' % (name, (end - start))

if __name__=='__main__':
    print 'Parent process %s.' % os.getpid()
    p = Pool()
    for i in range(5):
        p.apply_async(long_time_task, args=(i,))
    print 'Waiting for all subprocesses done...'
    p.close()
    p.join()
    print 'All subprocesses done.'

结果:

Parent process 669.
Waiting for all subprocesses done...
Run task 0 (671)...
Run task 1 (672)...
Run task 2 (673)...
Run task 3 (674)...
Task 2 runs 0.14 seconds.
Run task 4 (673)...
Task 1 runs 0.27 seconds.
Task 3 runs 0.86 seconds.
Task 0 runs 1.41 seconds.
Task 4 runs 1.91 seconds.
All subprocesses done.

进程间通讯

不同进程间可以通过Queue,Pipe来通信.Pipe用于两个进程间通信,Quene用于多个进程间通信.在只有两个进程通信的情况下Pipe效率高于Queue.

Pipe

导入Pipe模块:

from multiprocessing import Pipe

创建Pipe通信的两端(返回一个双元素的list):

p = Pipe(duplex=False)
其中duplex=False表示该Pipe只能单向通信.默认不写该参数为双向通信.

p[0],p[1]可以分别作为两个子进程的参数传递给子进程函数.也可以只传递一端给子进程,另一端交给父进程.

Pipe的两端可通过p.send()传送值,p.recv()接收值.

例子1:

from multiprocessing import Process, Pipe

def f(conn):
    conn.send([42, None, 'hello'])
    conn.close()

if __name__ == '__main__':
    parent_conn, child_conn = Pipe()
    p = Process(target=f, args=(child_conn,))
    p.start()
    print parent_conn.recv()   # prints "[42, None, 'hello']"
    p.join()

例子2:

import multiprocessing as mul

def proc1(pipe):
    pipe.send('hello')
    print('proc1 rec:',pipe.recv())

def proc2(pipe):
    print('proc2 rec:',pipe.recv())
    pipe.send('hello, too')


# Build a pipe

pipe = mul.Pipe()


# Pass an end of the pipe to process 1

p1   = mul.Process(target=proc1, args=(pipe[0],))

# Pass the other end of the pipe to process 2

p2   = mul.Process(target=proc2, args=(pipe[1],))
p1.start()
p2.start()
p1.join()
p2.join()

Queue

导入Queue模块:

from multiprocessing import Queue

创建Queue对象:

q = Queue(max)
其中max表示对象中可以存放的最大数量.

q可作为全局变量使用,也可以作为参数传递给子进程.
使用q.put()在Queue对象中放入需传递的值,q.get()取出值.

例子1:

from multiprocessing import Process,Queue

def writer_proc():
   q.put(100)

def reader_proc():
   print q.get()

if __name__ == '__main__':
    q = Queue()
    reader = Process(target=reader_proc,args=(q,))
    reader.start()
    writer = Process(target=writer_proc,args=(q,))
    writer.start()
    reader.join()
    writer.join()

例子2:

import multiprocessing

q = multiprocessing.Queue()

def reader_proc():
    print q.get()

reader = multiprocessing.Process(target=reader_proc)
reader.start()

q.put(100)
reader.join()

多线程

多任务除了使用多进程外还可以使用多线程来完成.单个进程中可以有多个线程,它们共享进程中的数据.

python中可使用高级模块Threading来创建多线程.其使用方法与multiprocessing相似.

导入Threading模块:

import Threading

*threading 模块提供的常用方法：
threading.currentThread(): 返回当前的线程变量。 (也可以使用threading.current_thread())
threading.enumerate(): 返回一个包含正在运行的线程的list。正在运行指线程启动后、结束前，不包括启动前和终止后的线程。
threading.activeCount(): 返回正在运行的线程数量，与len(threading.enumerate())有相同的结果。
threading模块提供的类：
Thread, Lock, Rlock, Condition, [Bounded]Semaphore, Event, Timer, local.*

构建新线程实例:

t = Threading.thread(target=function,...)
同时构建实例支持以下几种方法:

*Thread(group=None, target=None, name=None, args=(), kwargs={})
group: 线程组，目前还没有实现，库引用中提示必须是None；
target: 要执行的方法；
name: 线程名；
args/kwargs: 要传入方法的参数。*

实例支持以下方法:
*isAlive(): 返回线程是否在运行。正在运行指启动后、终止前。
get/setName(name): 获取/设置线程名。
is/setDaemon(bool): 获取/设置是否守护线程。初始值从创建该线程的线程继承。当没有非守护线程仍在运行时，程序将终止。
start(): 启动线程。
join([timeout]): 阻塞当前上下文环境的线程，直到调用此方法的线程终止或到达指定的timeout（可选参数）。*

使用:

t.start()
运行新线程.

如需等待线程运行结束:

t.join()

锁
由于多线程共享进程中的变量, 如果直接使用多线程修改变量的话容易出问题.所以多线程中一般会创建锁.

创建锁:

lock = threading.Lock()
此时有了一个锁的实例.

*锁的实例方法:
acquire([timeout]): 使线程进入同步阻塞状态，尝试获得锁定。
release(): 释放锁。使用前线程必须已获得锁定，否则将抛出异常。*

在每个线程需要修改变量前调用实例方法,尝试将修改变量的过程置于锁中,不让其他线程修改变量:

lock.acquire()

修改之后需要释放锁:

lock.release()

例子1:

balance = 0
lock = threading.Lock()

def run_thread(n):
    for i in range(100000):
        # 先要获取锁:
        lock.acquire()
        try:
            # 放心地改吧:
            change_it(n)
        finally:
            # 改完了一定要释放锁:
            lock.release()

例子2:

# encoding: UTF-8

import threading
import time

data = 0
lock = threading.Lock()

def func():
    global data
    print '%s acquire lock...' % threading.currentThread().getName()

    # 调用acquire([timeout])时，线程将一直阻塞，
    # 直到获得锁定或者直到timeout秒后（timeout参数可选）。
    # 返回是否获得锁。
    if lock.acquire():
        print '%s get the lock.' % threading.currentThread().getName()
        data += 1
        time.sleep(2)
        print '%s release lock...' % threading.currentThread().getName()

        # 调用release()将释放锁。
        lock.release()

t1 = threading.Thread(target=func)
t2 = threading.Thread(target=func)
t3 = threading.Thread(target=func)
t1.start()
t2.start()
t3.start()

其他锁及线程间通信见参考资料5中.

多线程的全局变量与局部变量
多线程之间修改全局变量需要加锁. 在线程的函数中创建局部变量可以解决加锁问题, 但如果线程需要运行不同函数, 函数之间需要共享变量, 局部变量调用不是很方便. threading.local()可以解决这个问题.
localschool = threading.local()

创建实例后, 不同线程在同时使用实例时不会产生冲突.

例子:

import threading


# 创建全局ThreadLocal对象:

local_school = threading.local()

def process_student():
    print 'Hello, %s (in %s)' % (local_school.student, threading.current_thread().name)

def process_thread(name):
    # 绑定ThreadLocal的student:
    local_school.student = name
    process_student()

t1 = threading.Thread(target= process_thread, args=('Alice',), name='Thread-A')
t2 = threading.Thread(target= process_thread, args=('Bob',), name='Thread-B')
t1.start()
t2.start()
t1.join()
t2.join()

分布式进程(待完善

协程

协程是单线程在不同的函数间中断并相互切换的一种运行模式.比如在函数A运行遇到阻塞时转向运行函数B,等到函数B运行结束再回来接着运行函数A.与多线程相比协程没有锁的问题.协程可以在IO密集的程序中节省IO等待时间,提高运行效率.

yield
python的生成器yield一定程度上支持协程.定义生成器yield可以直接在函数定义中将return换成yield.在调用生成器函数时首先将生成器赋给变量,通过变量的.next()方法调用生成器生成第一个值.再次调用.next()方法可生成第二个值..send(value)方法可在调用生成器时给它传递一个参数.
例子:

import time

def consumer():
    r = ''
    while True:
        n = yield r
        if not n:
            return
        print('[CONSUMER] Consuming %s...' % n)
        time.sleep(1)
        r = '200 OK'

def produce(c):
    c.next()
    n = 0
    while n < 5:
        n = n + 1
        print('[PRODUCER] Producing %s...' % n)
        r = c.send(n)
        print('[PRODUCER] Consumer return: %s' % r)
    c.close()

if __name__=='__main__':
    c = consumer()
    produce(c)

结果:

[PRODUCER] Producing 1...
[CONSUMER] Consuming 1...
[PRODUCER] Consumer return: 200 OK
[PRODUCER] Producing 2...
[CONSUMER] Consuming 2...
[PRODUCER] Consumer return: 200 OK
[PRODUCER] Producing 3...
[CONSUMER] Consuming 3...
[PRODUCER] Consumer return: 200 OK
[PRODUCER] Producing 4...
[CONSUMER] Consuming 4...
[PRODUCER] Consumer return: 200 OK
[PRODUCER] Producing 5...
[CONSUMER] Consuming 5...
[PRODUCER] Consumer return: 200 OK

更多关于yield见参考资料6.

gevent
gevent模块为python提供了完整的协程实现.

使用需先导入模块:

import gevent

在网络通信中一般还会导入monkey模块以将默认socket替换为可协程的socket:

from gevent import monkey
monkey.patch_socket()

或者将所有阻塞式调用,包括socket, ssl, threading, select都替换为异步式:

from gevent import monkey
monkey.patch_all()

这种替换调用一般放在第一行.替换后这些调用会自动处理阻塞问题.

在需使用协程时要用:

g = gevent.spawn(function,parament)
使用协程方式启动函数.参数为函数的参数.

等待任务结束可以使用:

g.join()

等待所有任务结束可以使用:

gevent.joinall(spawnlist)

不过使用monkey模块补丁自动处理有时候不能满足要求.这时我们可以使用其他模块.
如自动处理会并发所有连接,如果需要限制并发数量的话可以使用Pool模块.

from gevent.pool import Pool

新建一个Pool池:

p = Pool(number)
number为最高并发数

在并发池中启动函数:

p.spawn(function,parament)

等待所有任务结束:

p.join()

需要直接指定跳转时用sleep函数:

gevent.sleep(time)
其中time表示此处至少要阻塞time秒.

参考资料:

<<多进程>>
http://www.liaoxuefeng.com/wiki/001374738125095c955c1e6d8bb493182103fac9270762a000/0013868323401155ceb3db1e2044f80b974b469eb06cb43000

<>
http://www.coder4.com/archives/3352

<>
http://www.cnblogs.com/vamei/archive/2012/10/12/2721484.html

<>
https://blog.weizhe.net/?p=77

<>
http://www.cnblogs.com/huxi/archive/2010/06/26/1765808.html

<<生成器>>
http://wiki.jikexueyuan.com/project/start-learning-python/215.html

<> http://www.gevent.org/intro.html#installation-and-requirements