multiprocessing多进程常用方法总结

进程线程的区别

1.线程比进程调度速度更快

进程是资源分配的最小单位，所以如果创建多个进程，调度的时候进程之间的切换就伴随着资源的释放和重新分配，这样它的速度就慢很多。

而不同的线程之间是共享资源的，所以一个线程的结束，它不需要有资源的分配调度过程，直接就可以再起另外一个线程，这就是线程比进程调度速度更快的原因

2.进程能够利用多核

python是使用单核的工具，如果你使用线程顶多只能跑满一个核心，但我们的计算机现在都是多核处理器，所以只使用线程的利用率是比较低的。这种情况就可以使用进程，充分利用多个核心的优势

两者也可以搭配使用，先开进程再起线程

后起之秀：协程

接下来从代码来理解multiprocessing进程库

单个进程：

import multiprocessing 
import time 
def worker(interval):
    n= 5
    while n>0: 
        print("The time is {0}". format(time.ctime()))
        time.sleep(interval)
        n -= 1
if __name__ == "__main__": 
    p = multiprocessing.Process(target = worker, args=(3,))
    p.start()
    print("p.pid:",p.pid) 
    print("p.name:",p.name) 
    print("p.is_alive:",p.is_alive())

在p.start()开始时已经开启了一个子进程，但是程序的输出确实先将main函数的print打印出，再执行的子进程

多个进程

import multiprocessing 
import time 

def worker_1(interval):
    print('worker_1')
    time.sleep(interval)
    print("end worker_1")
    
def worker_2(interval):
    print('worker_2')
    time.sleep(interval)
    print("end worker_2")
    
def worker_3(interval):
    print('worker_3')
    time.sleep(interval)
    print("end worker_3")
    
if __name__ == "__main__": 
    p1 = multiprocessing.Process(target = worker_1, args=(2,))#生成p1对象时就已经执行到了worker_1函数了
    p2 = multiprocessing.Process(target = worker_2, args=(3,))
    p3 = multiprocessing.Process(target = worker_3, args=(4,))
    p1.start()
    p2.start()
    p3.start()
    
    print("The number of CPU is:" + str(multiprocessing.cpu_count()))#结果为4，是有一个主进程和三个子进程p1、p2、p3
    for p in multiprocessing.active_children():
        print("child p.name:" + p.name +"\tp.id:" + str(p.pid))
    print("End!!!!!!!!!!")

daemon属性

import multiprocessing 
import time 
def worker(interval):
    print('worker start:{0}'.format(time.ctime()))
    time.sleep(interval)
    print('worker end:{0}'.format(time.ctime()))
    
if __name__ == "__main__": 
    p = multiprocessing.Process(target = worker, args=(3,))
    #p.daemon = True
    p.start()
    print("main end!!")
    #因为子进程设置了daemon属性，所以主进程结束了，他们就随之结束了

上图第一次运行的是将注释取消，开启子进程的daemon属性的，第二个是没开启的。

join属性

import multiprocessing 
import time 
def worker(interval):
    print('worker start:{0}'.format(time.ctime()))
    time.sleep(interval)
    print('worker end:{0}'.format(time.ctime()))
    
if __name__ == "__main__": 
    p = multiprocessing.Process(target = worker, args=(3,))
    p.daemon = True
    p.start()
    p.join()#join属性，意味着告诉主进程，子进程执行完再执行主进程
    print("main end!!")

Lock锁

#当多个进程要访问共享资源的时候，Lock可以用来避免访问的冲突。当对文件进行多进程操作时，经常出现问题，所以需要使用Lock锁

#当多个进程要访问共享资源的时候，Lock可以用来避免访问的冲突。
import multiprocessing 
import sys

def worker_with(lock , f):
    #lock的第一种写法，推荐
    with lock:
        fs = open(f,'a+')
        n = 10
        while n>1:
            fs.write("Lockd acquire via with\n")
            n -= 1
        fs.close()
        
def workr_no_with(lock , f):
    #lock的第二种写法
    lock.acquire()
    try:
        fs = open(f,'a+')
        n = 10
        while n>1:
            fs.write("Lockd acquire directly\n")
            n -= 1
        fs.close()
    finally:
        lock.release()
        
if __name__ == "__main__": 
    lock = multiprocessing.Lock()
    f = "file.txt"
    w = multiprocessing.Process(target = worker_with, args=(lock,f))
    nw = multiprocessing.Process(target = workr_no_with, args=(lock,f))
    w.start()
    nw.start()
    print("main end!!!")

文本中的内容很整齐的写入了，但是如果不使用锁的话，当内容输入较多的时候，内容就会窜出现乱序的情况。

Semaphore属性用来控制对共享资源的访问数量，例如池的最大连接数

#semaphore用来控制对共享资源的访问数量，例如池的最大连接数
import multiprocessing 
import time

def worker(s,i):
    s.acquire()
    print(multiprocessing.current_process().name + " acquire")
    time.sleep(i)
    print(multiprocessing.current_process().name + " release\n")
    s.release()
    
if __name__ == "__main__": 
    s = multiprocessing.Semaphore(2)
    for i in range(5):
        p = multiprocessing.Process(target = worker,args = (s,i*2))
        p.start()

上述代码通过for循坏开启了5个子进程，但是因为定义了Semaphore(2)，所以一次只能有两个进程去运行，下图分别是不定义Semaphore与定义Semaphore的区别，结果如下：

queue属性，设置队列

'''
Queue是多进程安全的队列，可以使用Queue实现多进程之间的微据传递。put方法用以插入数据到队列中，put方选还有两个可造参数：blocked和timeout。
如果blocked为True（默认值），并且tineout为正维，该方法会阻塞timeout指定的时间，直到流从列有剩余的空间。如果超时，会抛出Queue.Fu11异常。
如果blocked为False，但该Queue已满，会立即抛出Queue.Fu11异常。get方法可以从队列读取并且除一个元素。
同样，get方法有两个可选参数；blocked和timeout。
如果blocked为True（默认值），并且tineout为正值，那么在等待时间内没有取到任何元龈，会性出Queue.Empty异常
如果blocked为False，有两种情况存在，如果Queue有一个值可用，则立即速露镇值，否则，如果队列为空，则立即抛出Queue.Empty养常。Queoe的一段示例代码
'''
import multiprocessing 
import time

def writer_proc(q):
    try:
        q.put(1,block = False)
    except:
        pass
       
def reader_proc(q):
    time.sleep(3)
    try:
        print(q.get(block = False))#队列中没有数据时如果不加block属性就会一直处于等待状态，加上block=False则队列中没有属性的时候会报错
    except:
        pass

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

代码中的注释已经对内容进行了解释，进程先调用到了writer_proc函数向队列中写入一个数，然后reader_proc函数中将队列内容取出

Pipe方法

'''
Pipe方法返回(conn1，conn2)代表一个管道的两个端。
Pipe方法有dupLex参数，如果duplex参数为True(默认值)。那么这个管道是全双工模式，也就是说conn1和conn2均可收发。
duplex为False，conn1只负责接受消息，conn2只负责发送消息。send和recv方法分别是发送和接受消息的方法。
例如，在全双工模式下，可以调用connl.send发送消息，connl.recv接收消息。如果没有消息可接收，recv方法会一直阻塞。如果管道已经被关闭，那么recv方法会抛出EOFError。
'''
import multiprocessing 
import time

def proc1(pipe):
    while True:
        for i in range(1000):
            print("proc1 send:%s"%(i))
            pipe.send(i)
            time.sleep(1)
            
def proc2(pipe):
    while True:
        print("proc2 rev:",pipe.recv())
        time.sleep(1)
        
def proc3(pipe):
    while True:
        print("proc3 rev:",pipe.recv())
        time.sleep(1)

if __name__ == "__main__": 
    pipe = multiprocessing.Pipe()
    print(pipe[0])
    print(pipe[1])
    p1 = multiprocessing.Process(target = proc1, args=(pipe[0],))
    p2 = multiprocessing.Process(target = proc2, args=(pipe[1],))
    p3 = multiprocessing.Process(target = proc3, args=(pipe[1],))
    p1.start()
    p2.start()
    p3.start()
    p1.join()
    p2.join()
    p3.join()

当开启三个进程两个接收信息，一个发送的时候，接收段会交替接收

进程池Pool

#在利用Python进行系统管理的时候，特别是同时操作多个文件目录，或看远程控制多台主机，并行操作可以节约大量的时间。当被操作对象数目不大时，可以重接利用muLtiprocessing中的Proce
import multiprocessing
import time

def func(msg):
    print("msg:",msg)
    time.sleep(3)
    print("end")

#非阻塞
if __name__ == "__main__": 
    pool = multiprocessing.Pool(processes = 3)#定义三个进程在进程池中
    '''
    apply_async(func[,args[,kwds[,callbackl]]])它是非阻塞,apply(func[,args[,kwds]])是阻塞的（理解区别，看例1例2结果区别）
    close()关闭pool，使其不在接受新的任务。
    teminate()结束工作进程。不在处理未完成的任务。
    join()主进程阻塞，等待子进程的退出，join方法要在close或terminate之后使用。
    '''
    for i in range(4):
        msg = "hello %d" %(i)
        pool.apply_async(func,(msg,))#维持执行的进程总数为processes，当一个进程执行完毕后会添加新的进程进去
    print('ohhhhhhhhhhhhhhh')
    pool.close()
    pool.join()#调用join之前，先调用close函数，否则会出错。执行完close后不会有新的进程加入到pool，join函数等待所有子进程结束
    print("Sub-process(es) done")

原本定义了进程池中有三个进程，但for循环开了四个进程，所以先有三个进程运行，等待一个结束，第四个进程才会再运行，注意使用apply_async这种非阻塞的写法，效率更高