Python之多进程多线程

一、多进程与多线程的概念

　　1.多进程的概念

　　进程是程序在计算机上的的一次执行活动。当你运行一个程序，你就启动了一个进程。显然，程序是死的（静态的），进程是活的（动态的）。进程可以分为系统进程和用户进程。凡是用于完成操作系统的各种功能的进程就是系统进程，它们就是处于运行状态下的操作系统本身；由用户本身启动的进程都是用户进程。进程是操作系统进行资源分配的单位。

　　在操作系统的管理下，所有正在运行的进程轮流使用CPU，每个进程运行占用CPU时间非常短，用户根本感觉不出来CPU是在轮流为多个进程服务，就好像所有的进程都在不间断地运行一样。但实际上在任何一个时间内有且仅有一个进程占有CPU。

　　2.多线程的概念

　　　每个正在系统上运行的程序都是一个进程。每个进程包含一到多个线程。进程也可能是整个程序或者是部分程序的动态执行。线程是一组指令的集合，或者是程序的特殊段，它可以在程序里独立执行。也可以把它理解为代码运行的上下文。所以线程基本上是轻量级的进程，它负责在单个程序里执行多任务。通常由操作系统负责多个线程的调度和执行。线程是程序中一个单一的顺序控制流程.在单个程序中同时运行多个线程完成不同的工作,称为多线程.多线程是为了同步完成多项任务，不是为了提高运行效率，而是为了提高资源使用效率来提高系统的效率。线程是在同一时间需要完成多项任务的时候实现的。

　　3.多进程与多线程的区别：

　　多进程使用的是CPU的一个核，适合IO密集型

　　多线程使用的是CPU的多个核，适合运算密集型

二、多进程实例

　　创建一个进程(Process)对象:

　　p = multiprocessing.Process(target=worker 1, args=(2,))

　　target = 函数名字

　　args = 函数需要的参数，以tuple的形式传入

　　注意：单个元素的tuple的表现形式

　　multiprocessing用到的两个方法：

　　cpu_count()     统计CPU总数

　　active_children()    获得所有子进程

　　实例一：程序代码如下：

import multiprocessing

import time


def worker(interval):
    time.sleep(interval)
    print ("hello world")

if __name__ == "__main__":
    p = multiprocessing.Process(target=worker, args=(5,))
    p.start()
    print (p.is_alive())
    p.join(timeout = 3)    #等待子进程执行完毕或超时退出
    print ("end main")
    print (p.name)
    print (p.pid)

　　结果：

　　实例二：程序代码如下：

import multiprocessing

import time


def worker(name,interval):
    print ("{0} start".format(name))
    time.sleep(interval)
    print ("{0} end".format(name))

if __name__ == "__main__":
    print ("main start")
    print ("this computer has {0}".format(multiprocessing.cpu_count()))
    p1 = multiprocessing.Process(target=worker,args=("worker1",2))
    p2 = multiprocessing.Process(target=worker,args=("worker2",3))
    p3 = multiprocessing.Process(target=worker,args=("worker3",4))
    p1.start()
    p2.start()
    p3.start()
    for p in multiprocessing.active_children():
        print ("the pid of {0} is {1}".format(p.name,p.pid))
    print ("main end")

　　结果：

　　

三、多进程锁

　　Lock组件：当我们用多进程来读写文件的时候，如果一个进程是写文件，一个进程是读文件，如果两个文件同时进行，肯定是不行的，必须是文件写结束以后，才可以进行读操作。或者多进程在共享一些资源的时候，同时只能有一个进程进行访问，那么就要有一个锁机制进行控制。当多个进程需要访问共享资源的时候，Lock可以用来避免访问的冲突。

　　lock.acquire()：获取锁

　　lock.release()：释放锁

　　1.使用acquire与release方法：

import multiprocessing

import time


def add(number,value,lock):
    lock.acquire()
    try:
        print ("init add{0} number = {1}".format(value, number))
        for i in xrange(1, 6):
            number += value
            time.sleep(1)
            print ("add{0} number = {1}".format(value, number))
    except Exception as e:
        raise e
    finally:
        lock.release()

if __name__ == "__main__":
    lock = multiprocessing.Lock()
    number = 0
    p1 = multiprocessing.Process(target=add,args=(number,1,lock))
    p2 = multiprocessing.Process(target=add,args=(number,3,lock))
    p1.start()
    p2.start()
    print ("main end")

　　结果：

　　2.使用with lock

import multiprocessing

import time


def add(number,value,lock):
    with lock:
        print ("init add{0} number = {1}".format(value,number))
        for i in xrange(1,6):
            number += value
            time.sleep(1)
            print ("add{0} number = {1}".format(value,number))
if __name__ == "__main__":
    lock = multiprocessing.Lock()
    number = 0
    p1 = multiprocessing.Process(target=add,args=(number,1,lock))
    p2 = multiprocessing.Process(target=add,args=(number,3,lock))
    p1.start()
    p2.start()
    print ("main end")

　　结果：

四、多进程共享内存

　　共享内存（Shared Memory）是最简单的进程间通信方式，它允许多个进程访问相同的内存，一个进程改变其中的数据后，其他的进程都可以看到数据的变化。

　　共享内存是进程间最快速的通信方式：

• 进程共享同一块内存空间。
• 访问共享内存和访问私有内存一样快。
• 不需要系统调用和内核入口。
• 不造成不必要的内存复制。

　　内核不对共享内存的访问进行同步，因此程序员必须自己提供同步。
　　使用共享内存：

• 某个进程分配内存段。
• 使用这个内存段的进程要连接（attach）这个内存段。
• 每个进程使用完共享内存段后，要分离（detach）这个内存段。
• 在某个地方，必须有一个进程来销毁这个内存段。

　　python的共享内存操作由multiprocessing模块提供，multiprocessing提供了Value和Array模块，可以在不通的进程中共同使用。程序例子如下：

　　

import multiprocessing

import time
def add(number,add_value,lock):
    lock.acquire()
    try:
        print ("init add{0} number = {1}".format(add_value, number.value))
        for i in xrange(1, 6):
            number.value += add_value
            print ("--------add{0} has added----------".format(add_value))
            print ("
")
            time.sleep(1)
            print ("add{0} number = {1}".format(add_value, number.value))
    except Exception as e:
        raise e
    finally:
        lock.release()
if __name__ == "__main__":
    lock = multiprocessing.Lock()
    number = multiprocessing.Value("i",0)
    p1 = multiprocessing.Process(target=add, args=(number, 1,lock))
    p2 = multiprocessing.Process(target=add, args=(number, 3,lock))
    p1.start()
    p2.start()
    print ("main end")

结果：

main end
init add1 number = 0
--------add1 has added----------


add1 number = 1
--------add1 has added----------


add1 number = 2
--------add1 has added----------


add1 number = 3
--------add1 has added----------


add1 number = 4
--------add1 has added----------


add1 number = 5
init add3 number = 5
--------add3 has added----------


add3 number = 8
--------add3 has added----------


add3 number = 11
--------add3 has added----------


add3 number = 14
--------add3 has added----------


add3 number = 17
--------add3 has added----------


add3 number = 20

　　以上实现的数据共享的方式只有两种结构value和Array。Python中提供了更强大的manager专门用来做数据分享，其支持的类型非常多，包括Value、array、list、dict、Queue、lock等。实例如下：

import multiprocessing

def worker(d,l):
    l += range(11,16)
    for i in xrange(1,6):
        key = "key{0}".format(i)
        val = "val{0}".format(i)
        d[key] = val


if __name__ == "__main__":
    manager = multiprocessing.Manager()
    d = manager.dict()
    l = manager.list()
    p = multiprocessing.Process(target= worker,args=(d,l))
    p.start()
    p.join()
    print (d)
    print (l)
    print ("main end ")

　　结果：

五、进程池

　　Pool可以提供指定数量的进程供用户调用，当有新的请求提交到pool中时，如果池还没有满，那么就会创建一个新的进程用来执行该请求；但如果池中的进程数已经达到规定最大值，那么该请求就会等待，直到池中有进程结束，才会创建新的进程来它。相关函数：

• apply_async(func[, args[, kwds[, callback]]]) 它是非阻塞
• apply(func[, args[, kwds]])是阻塞的
• close()    关闭pool，使其不在接受新的任务
• join()    主进程阻塞，等待子进程的退出， join方法要在close使用。

　　非阻塞实例如下：

import multiprocessing

import time


def worker(msg):
    print ("########start {0}#########".format(msg))
    time.sleep(1)
    print ("########end {0}#########".format(msg))
if __name__ == "__main__":
    print ("main start")
    pool = multiprocessing.Pool(processes=3)      #创建进程池
    for i in xrange(1,10):
        msg = "hello {0}".format(i)
        pool.apply_async(func= worker ,args=(msg,))     #非阻塞

    pool.close()
    pool.join()    #在join之前，一定要调用close，否则报错
    print ("main end")

结果：

　　阻塞实例如下：

import multiprocessing

import time


def worker(msg):
    print ("########start {0}#########".format(msg))
    time.sleep(1)
    print ("########end {0}#########".format(msg))
if __name__ == "__main__":
    print ("main start")
    pool = multiprocessing.Pool(processes=3)      #创建进程池
    for i in xrange(1,10):
        msg = "hello {0}".format(i)
        pool.apply(func= worker ,args=(msg,))     #非阻塞

    pool.close()
    pool.join()    #在join之前，一定要调用close，否则报错
    print ("main end")

结果：

 

六、线程实例

　　多线程有两种实现方法：

　　1.将要执行的方法作为参数传给Thread的构造方法： t = threading.Thread(target = action, agrs =(i,))

　　2.从Thread继承，重写run()

　　实例代码如下：

import threading


def worker(n):
    print ("start worker{0}".format(n))

class MyThread(threading.Thread):
    def __init__(self,args):
        super(MyThread,self).__init__()
        self.args = args
    def run(self):
        print ("start MyThread{0}".format(self.args))

if __name__ == "__main__":
    for i in xrange(1,6):
        t1 = threading.Thread(target= worker,args=(i,))
        t1.start()
    t1.join()
    for x in xrange(6,11):
        t2 = MyThread(x)
        t2.start()
    t2.join()

结果：

start worker1
start worker2
start worker3
start worker4
start worker5
start MyThread6
start MyThread7
start MyThread8
start MyThread9
start MyThread10

七、多线程锁

　　通过threading.lock()来创建锁，函数在执行的只有先要获得锁，然后执行完以后要释放锁。获得锁与释放锁的方法与进程锁相同

　　程序实例如下：

import threading
import time


def worker(name,lock):
    with lock:
        print ("start {0}".format(name))
        time.sleep(5)
        print ("end {0}".format(name))

#with lock与lock.acquire(),lock.release()
if __name__ == "__main__":
    lock = threading.Lock()
    t1 = threading.Thread(target=worker,args=("worker1",lock))
    t2 = threading.Thread(target=worker,args=("worker2",lock))
    t1.start()
    t2.start()
    print ("main end")

结果：

start worker1
main end
end worker1
start worker2
end worker2

八、线程池

　　多线程和多进程的不同之处在于多线程本身就是可以和父进程共享内存的，这也是为什么其中一个线程挂掉以后，为什么其它线程也会死掉的道理。程序实例如下：

import threading


def worker(l):
    l.append("ling")
    l.append("huo")
    l.append("wang")

if __name__ == "__main__":
    l = list()
    l += range(1, 10)
    print (l)
    t = threading.Thread(target=worker,args=(l,))
    t.start()
    print (l)

结果：

[1, 2, 3, 4, 5, 6, 7, 8, 9]
[1, 2, 3, 4, 5, 6, 7, 8, 9, 'ling', 'huo', 'wang']

九、线程池

　　首先需要安装包：pip  install threadpool。语法格式：

pool = ThreadPool(poolsize)  
requests = makeRequests(some_callable, list_of_args, callback)  
[pool.putRequest(req) for req in requests]  
pool.wait()  

　　第一行定义了一个线程池，表示最多可以创建poolsize这么多线程；

　　第二行是调用makeRequests创建了要开启多线程的函数，以及函数相关参数和回调函数，其中回调函数可以不写，default是无，也就是说makeRequests只需要2个参数就可以运行；

　　第三行用法比较奇怪，是将所有要运行多线程的请求扔进线程池，[pool.putRequest(req) for req in requests]等同于

　　for req in requests:  

　　　　 pool.putRequest(req) 

　　第四行是等待所有的线程完成工作后退出。