python 系列之 - 多进程

桃扇骨 2022-08-21 00:26 180阅读 0赞

**多进程 (multiprocessing)**

1 创建一个多进程  
**创建：** multiprocessing.Process(\[target=函数名\], \[ name = 别名\], \[args=(参数，)\], \[kwargs=调用对象的字典,\])  
**方法：**  
is\_alive() ： 判断当前进程是否为活动状态  
join(\[timeout\])： 和多线程一样，等待其它子进程结束后主进程才继续执行  
run()： 调用start()方法后自动调用run()方法  
start()： Process以start()启动某个进程。  
terminate()： 结束进程方法  
**属性：**  
daemon： 设置子进程为后台进程，默认为前台进程（False) ,父进程终止后自动终止，且自己不能产生新进程，必须在start()之前设置。  
pid: 获取进程的ID号  
name： 获取进程的别名  
exitcode： 进程在运行时为None、如果为–N，表示被信号N结束

import multiprocessing
    import time
    def do_process(num):
        print("func in process %s" % num)
    
    if __name__ == "__main__":
        for i in range(3):
            p = multiprocessing.Process(target=do_process, name="My_name-{0}".format(str(i)), args=(str(i), ))
            p.start()
            print("Process {0} is started, PID = {1}, is_alive = {2} ".format(p.name, p.pid, p.is_alive()))
    
    run result:
    Process My_name-0 is started, PID = 3440, is_alive = True 
    func in process 0
    Process My_name-1 is started, PID = 11904, is_alive = True 
    func in process 1
    Process My_name-2 is started, PID = 24396, is_alive = True 
    func in process 2

自定义进程类

import multiprocessing
    import time
    
    class ClockProcess(multiprocessing.Process):
        def __init__(self, interval):
            super(ClockProcess, self).__init__()
            self.interval = interval
    
        def run(self):
            n = 5
            while n > 0:
                print("the time is {0}".format(time.ctime()))
                time.sleep(self.interval)
                n -= 1
    
    if __name__ == '__main__':
        p = ClockProcess(3)
        p.start()
    
    注：进程p调用start()时，自动调用run()    
    run result：
    the time is Tue Mar  8 14:34:21 2016
    the time is Tue Mar  8 14:34:24 2016
    the time is Tue Mar  8 14:34:27 2016
    the time is Tue Mar  8 14:34:30 2016
    the time is Tue Mar  8 14:34:33 2016

**2 进程间数据共享**  
**1） Queue**  
Queue是多进程安全的队列，可以使用Queue实现多进程之间的数据传递。  
**put(obj, block=True, timeout=None)** 方法用以插入数据到队列中，put方法还有两个可选参数：blocked和timeout。  
如果blocked为True（默认值），无timeout参数，则阻塞，直到取走后再继续put  
如果blocked为True，并且timeout为正值，当队列满了之后，该方法会阻塞timeout 的时间，直到该队列有剩余的空间。如果超时，队列仍满着，会抛出Queue.Full异常。  
如果blocked为False，但该Queue已满，会立即抛出Queue.Full异常,与 put\_nowait()方法等价

**get(block=True, timeout=None)**： 方法可以从队列读取并且删除一个元素。同样，get方法有两个可选参数：blocked和timeout。  
如果blocked为True（默认值），timeout无值，则阻塞等待有值以后再取  
如果blocked为True，并且timeout为正值，那么在等待时间内没有取到任何元素，会抛出Queue.Empty异常。  
如果blocked为False，有两种情况存在： 如果Queue有一个值可用，则立即返回该值，否则，如果队列为空，则立即抛出Queue.Empty异常

import multiprocessing
    from multiprocessing import Queue
    import time
    
    def dofunc(qname, num):
        qname.put(num)
        print("put %s to qname" % num)
        time.sleep(1)
    
    def get_queue(q1):
        value = q1.get()
        print("get value %s from q" % value)
        time.sleep(2)
    
    
    if __name__ == '__main__':
        q = Queue(5)
        for i in range(10):
            p = multiprocessing.Process(target=dofunc, args=(q, i))
            p.start()
        while 1:
            get_queue(q)
            if q.empty():
                break
        print("Finished")

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

import multiprocessing
    from multiprocessing import Queue, Pipe
    import time
    
    def pip_send(pip):
        for i in range(10):
            print("send to pip %s" % i)
            pip.send(i)
            time.sleep(0.5)
    
    def pip_recv(pip):
        while True:
            value = pip.recv()
            print("recv from pip %s" % value)
            slice(1)
            if not pip.recv():
                break
    
    if __name__ == "__main__":
        pip = Pipe()
        p1 = multiprocessing.Process(target=pip_send, args=(pip[0], ))
        p2 = multiprocessing.Process(target=pip_recv, args=(pip[1], ))
        p1.start()
        p2.start()
    
        p1.join()
        p2.join()

**3) Managers**  
Manager提供了一种不同进程间创建共享数据的方式，由Manager()创建来保存Python对象并允许其它进程来进行访问。支持的类型有 list, dict, Namespace, Lock, RLock, Semaphore, BoundedSemaphore, Condition, Event, Barrier, Queue, Value and Array

import multiprocessing
    import time
    
    def worker(d, key, value):
        d[key] = value
    
    if __name__ == '__main__':
        mgr = multiprocessing.Manager()
        d = mgr.dict()
        jobs = [multiprocessing.Process(target=worker, args=(d, i, i * 2)) for i in range(10)]
        print(jobs)
        for j in jobs:
            j.start()
        for j in jobs:
            j.join()
    
        print('Results:')
        print(d)
    
        for key, value in d.items():
            print("%s=%s" % (key, value))
    run result:
    {
       0: 0, 1: 2, 2: 4, 3: 6, 4: 8, 5: 10, 6: 12, 7: 14, 8: 16, 9: 18}
    0=0
    1=2
    2=4
    3=6
    4=8
    5=10
    6=12
    7=14
    8=16
    9=18

**4） Pool**  
在利用Python进行系统管理的时候，特别是同时操作多个文件目录，或者远程控制多台主机，并行操作可以节约大量的时间。  
当被操作对象数目不大时，可以直接利用multiprocessing中的Process动态成生多个进程，十几个还好，但如果是上百个，上千个目标，不加限制的开启进程恐怕服务器就该挂了，  
但我们手动的去限制进程数量却又太过繁琐，此时可以发挥进程池的功效。  
Pool可以提供指定数量的进程，供用户调用，当有新的请求提交到pool中时，如果池还没有满，那么就会创建一个新的进程用来执行该请求；但如果池中的进程数已经达到规定最大值，那么该请求就会等待，直到池中有进程结束，才会创建新的进程来它。

进程池方法：  
apply(func\[, args\[, kwds\]\])： 阻塞的执行，比如创建一个有3个线程的线程池，当执行时是创建完一个 执行完函数 在创建另一个，变成一个线性的执行  
apply\_async(func\[, args\[, kwds\[, callback\]\]\]) ： 它是非阻塞执行，同时创建3个线程的线城池，同时执行，只要有一个执行完立刻放回池子待下一个执行，并行的执行  
close()： 关闭pool，使其不在接受新的任务。  
terminate() ： 结束工作进程，不在处理未完成的任务。  
join() 主进程阻塞，等待子进程的退出， join方法要在close或terminate之后使用。

**非阻塞效果**

import multiprocessing
    import time
    
    def pool_func(i):
        print("start %s" %i)
        time.sleep(3)
        print("end - %s" %i)
    
    if __name__ == "__main__":
        pool = multiprocessing.Pool(3)
        for i in range(5):
            pool.apply_async(pool_func, args=(i,))  #Pool执行函数，apply执行函数,当有一个进程执行完毕后，会添加一个新的进程到pool中
        pool.close()
        pool.join() #调用join之前，一定要先调用close() 函数，否则会出错, close()执行后不会有新的进程加入到pool,join函数等待素有子进程结束
    run result：
    start 0
    start 1
    start 2
    end -  0
    start 3
    end -  1
    start 4
    end -  2
    end -  3
    end -  4

**阻塞效果：**

import multiprocessing
    import time
    
    def pool_func(i):
        print("start %s" %i)
        time.sleep(3)
        print("end - %s" %i)
    
    if __name__ == "__main__":
        pool = multiprocessing.Pool(3)
        for i in range(5):
            pool.apply(pool_func, args=(i,))
        pool.close()
        pool.join()
    run result：
    start 0
    end -  0
    start 1
    end -  1
    start 2
    end -  2
    start 3
    end -  3
    start 4
    end -  4