python 进程回顾

╰+攻爆jí腚メ 2021-11-15 14:00 298阅读 0赞

# 如何生成进程 #

import multiprocessing
    
    def foo(i):
        print("called function foo in process: %s" % i )
        process_name = multiprocessing.current_process().name
        print(process_name)
        return
    
    if __name__ == '__main__':
        process_job = []
        for i in range(5):
            p = multiprocessing.Process(target=foo,name=f"process_{i}",args=(i,))
            process_job.append(p)
            print("main:",p.name)
            p.start()
        for job in process_job:
            job.join() # 如果不json,子进程会空闲下来，但不会消失，需要我们自己kill掉

# 后台进程 #

p.daemon = True 
    """ p.daemon = True 必须在 p.start()之前设置 daemon=True 以为 后台运行，守护进程。 当主进程结束后，后台进程不管是否执行完毕都会随主进程而结束。 # 如果主进程没有json,会直接结束呢。 注意： 后台进程 不允许再创建子进程 """

# 杀死进程 #

p.terminate() # 立刻杀死一个进程
    p.is_alive()  # 是否存活

# 子类中使用进程 #

同多线程一样，略
    https://blog.csdn.net/sunt2018/article/details/97248207

# 如何在进程间 交换对象 #

## 使用队列进行交换对象 ##

对象必须为python可序列化的 对象 ，即可用`pickable`模块序列化的

import multiprocessing
    import random
    import time
    
    class producer(multiprocessing.Process):
        def __init__(self,queue):
            multiprocessing.Process.__init__(self)
            self.queue = queue
        def run(self):
            for i in range(10):
                item = random.randint(0,256) # 随机生成一个0-256的数
                self.queue.put(item)         # 放入队列
                print("producer: ",item,"appended to queue",self.name)
                time.sleep(1)
                print("queue size is",self.queue.qsize())
    
    
    class consumer(multiprocessing.Process):
        def __init__(self,queue):
            multiprocessing.Process.__init__(self)
            self.queue = queue
    
        def run(self) -> None:
            while True:
                if (self.queue.empty()):
                    print("queue is empty")
                    break
                else:
                    time.sleep(2)
                    item = self.queue.get()
                    print("consumer: ",item,"poped from by",self.name)
                    time.sleep(1)
    
    if __name__ == '__main__':
        queue = multiprocessing.Queue()
        process_producer = producer(queue)
        process_consumer = consumer(queue)
    
        process_producer.start()
        process_consumer.start()
    
        process_producer.join()
        process_consumer.join()

## 使用管道进行对象交换 ##

import multiprocessing
    
    def create_items(pipe):
        output_pipe , _  = pipe
        for item in range(10):
            output_pipe.send(item)
        output_pipe.close()
    
    def multiply_items(pipe_1,pipe_2):
        close,input_pipe = pipe_1
        close.close()
        output_pipe,_ = pipe_2
        try:
            while True:
                item = input_pipe.recv()
                output_pipe.send(item * item)
        except EOFError:
            output_pipe.close()
    
    if __name__ == '__main__':
        # 第一个管道 0-9 数字
        pipe_1 = multiprocessing.Pipe(True)
        process_pipe_1 = multiprocessing.Process(
                        target=create_items,args=(pipe_1,))
        process_pipe_1.start()
    
        # 第二个管道
        pipe_2 = multiprocessing.Pipe(True)
        process_pipe_2 = multiprocessing.Process(
            target=multiply_items, args=(pipe_1,pipe_2))
        process_pipe_2.start()
    
        pipe_1[0].close()
        pipe_2[0].close()
    
        try:
            while True:
                print(pipe_2[1].recv())
        except EOFError:
            print("END")
    
    """ multiprocessing.Pipe() 返回一个pari(conn1,conn2)，分别代表管道的两端。 如果参数duplex是True表示是双工。 如果是False单向的:conn1接收，conn2发送 send和recv方法分别是发送和接受消息的方法。例如，在全双工模式下，可以调用conn1.send发送消息，conn1.recv接收消息。 如果没有消息可接收，recv方法会一直阻塞。如果管道已经被关闭，那么recv方法会抛出EOFError。 """

# 如何同步进程状态 #

[https://www.cnblogs.com/dukuan/p/9810812.html][https_www.cnblogs.com_dukuan_p_9810812.html]

1、基本概念

多个进程可以协同工作来完成一项任务，通常需要共享数据。所以在多进程之间保持数据的一致性就很重要，需要共享数据协同的进程必须以适当的策略来读写数据。同步原语和线程的库类似。

- Lock：一个Lock对象有两个方法acquire和release来控制共享数据的读写权限。

- Event：一个进程发事件的信号，另一个进程等待事件的信号。Event对象有两个方法set和clear来管理自己内部的变量。

- Condition：此对象用来同步部分工作流程，在并行的进程中，有两个基本的方法，wait()用来等待进程，notify\_all用来通知所有等待此条件的进程。

- Semaphore：用来共享资源，比如：支持固定数据的共享连接。

- RLock：递归锁对象，其用途和方法同Threading模块一样。

- Barrier：将程序分成几个阶段，适用于有些进程必须在某些特性进程之后执行，处于Barrier之后的代码不能同处于Barrier之前的代码并行。

2、测试用例

使用barrier函数来同步两个进程,一共四个进程，分成两组对照试验。

import multiprocessing
    from multiprocessing import Barrier, Lock, Process
    from time import time
    from datetime import datetime
    
    def test_with_barrier(synchronizer, serializer):
        """ test_with_barrier函数调用了barrier的wait()方法， 当两个进程都调用wait()方法时，他们会一起继续执行。 """
        name = multiprocessing.current_process().name
        synchronizer.wait()
        now = time()
        with serializer:
            print("process %s ----> %s" %(name, datetime.fromtimestamp(now)))
    
    def test_without_barrier():
        name = multiprocessing.current_process().name
        now = time()
        print("process %s ----> %s" %(name, datetime.fromtimestamp(now)))
    
    if __name__ == "__main__":
        # create a barrier and lock.
        synchronizer = Barrier(2)
        serializer = Lock()
        
        # create four processes
        Process(name='p1 - test_with_barrier', 
                target=test_with_barrier, args=(synchronizer, serializer)).start()
        Process(name='p2 - test_with_barrier', 
                target=test_with_barrier, args=(synchronizer, serializer)).start()
        Process(name='p3 - test_without_barrier', 
                target=test_without_barrier).start()
        Process(name='p4 - test_without_barrier', 
                target=test_without_barrier).start()

# 如何管理进程间状态 #

进程之间管理状态

Python的多进程模块提供了在所有的用户间管理共享信息的管理者(Manager)，一个管理者对象控制着持有Python对象的服务进程，并允许其他进程操作共享对象。

管理者特性：

- 它控制着管理共享对象的服务进程

- 它确保当某一进程修改了共享对象之后，所有的进程拿到的共享对象都得到了更新。

import multiprocessing
    manager = multiprocessing.Manager()
    
    # dict 等等方法，进程间可以通过这些传递 共享变量。
    dictionary = manager.dict()

# 进程池 #

[https://www.cnblogs.com/zhangfengxian/p/python-process-pool.html][https_www.cnblogs.com_zhangfengxian_p_python-process-pool.html]

当需要创建的子进程数量不多时，可以直接利用multiprocessing中的Process动态成生多个进程，但如果是上百甚至上千个目标，手动的去创建进程的工作量巨大，此时就可以用到multiprocessing模块提供的Pool方法。

初始化Pool时，可以指定一个最大进程数，当有新的请求提交到Pool中时，如果池还没有满，那么就会创建一个新的进程用来执行该请求；但如果池中的进程数已经达到指定的最大值，那么该请求就会等待，直到池中有进程结束，才会用之前的进程来执行新的任务，请看下面的实例：

import multiprocessing
    import time
    import os
    import random
    
    
    def test1(msg):
            t_start = time.time()   
            print("%s开始执行，进程号为%d" % (msg, os.getpid()))
            time.sleep(random.random() * 2)  
            t_stop = time.time()
            print("%s执行完成，耗时%.2f" % (msg, t_stop - t_start)) 
    
    
    if __name__ == "__main__":
        
            po = multiprocessing.Pool(3)
            for i in range(0, 10):
                    # Pool().apply_async(要调用的目标,(传递给目标的参数元祖,))
                    # 每次循环将会用空闲出来的子进程去调用目标
                    po.apply_async(test1, (i,))
        
            print("-----start-----")
        
            po.close() # 关闭进程池，关闭后po不再接收新的请求
            po.join() # 等待po中所有子进程执行完成，必须放在close语句之后
    
            print("-----end-----")

multiprocessing.Pool常用函数解析：

*  apply\_async(func\[, args\[, kwds\]\]) ：使用非阻塞方式调用func（并行执行，堵塞方式必须等待上一个进程退出才能执行下一个进程），args为传递给func的参数列表，kwds为传递给func的关键字参数列表；
 *  close()：关闭Pool，使其不再接受新的任务；
 *  terminate()：不管任务是否完成，立即终止；
 *  join()：主进程阻塞，等待子进程的退出， 必须在close或terminate之后使用；

[https_www.cnblogs.com_dukuan_p_9810812.html]: https://www.cnblogs.com/dukuan/p/9810812.html
[https_www.cnblogs.com_zhangfengxian_p_python-process-pool.html]: https://www.cnblogs.com/zhangfengxian/p/python-process-pool.html