devbox
我家自动化
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Python基础之多进程_python多进程

作者：我家自动化 | 2024-07-21 11:24:42

赞

踩

python多进程

目录

1、什么是多进程

1.2 进程的状态

2、进程的创建-multiprocessing

2.1 Process类语法说明

2.2 进程pid

2.3 给子进程指定的函数传递参数

2.4 进程间不同享全局变量

3、 进程间同步-Queue

3.1 Queue类语法说明

3.2 Queue的使用

3.3 Queue实例

4、进程间同步-Lock

5、进程池Pool

5.2 Pool实例

5.3 进程池中的Queue

6、进程、线程对比

6.1功能

6.2 区别

6.3 优缺点

1、什么是多进程

一个程序运行起来，代码+用到的资源称之为进程，它是操作系统分配资源的基本单元。不仅可以通过线程完成多任务，进程也可以。

1.2 进程的状态

工作中，任务数往往大于cpu的核数，即一定有一些任务正在执行而另一些任务在等待cpu进行执行，就导致了有不同的状态。

就绪态

执行态

等待态

2、进程的创建-multiprocessing

2.1 Process类语法说明

multiprocessing模块通过创建一个Process对象然后调用它的start()方法来生成进程，Process与threading.ThreadAPI相同

语法格式：multiprocessing.Process(group=None,target=None.name=None,args=(),kwargs={},*,daemon=None)

参数说明：

group:指定进程组，大多数情况用不到

target:如果传递了函数的引用，可以认为这个子进程就执行这里的代码

name:给进程设定一个名字，可以不设定

args:给target指定的函数传递参数，以元组的方式传递

kwargs:给target指定的函数传递命名参数

multprocessing.Process对象具有如下方法和属性：

run()  # 进程具体执行的方法

start()  # 启动子进程实例

join[timeout]  # 如果可选参数timeout是默认值None,则将阻塞至调用join()方法的进程终止；如果timeout是一个正数，则最多会阻塞timeout秒

name  # 当前进程的别名，默认为Process-N,N为从1开始递增的整数

pid  # 当前进程的pid（进程号）

is_alive()  # 判断进程子进程是否还在活着

daemon  # 进程守护标志，是一个布尔值

authkey  # 进程的身份验证密钥

sentinel  # 系统对象的数字句柄，当进程结束时将会变成ready

timeinate()  # 不管任务是否完成，立即终止子进程

kill()  # 与timeimate()相同，但在unix上使用SIGKILL信号

close()  # 关闭Process对象，释放与之关联的所有资源

  1. from multiprocessing import Process
  2. import time
  3.  
  4.  
  5. def run_proc():
  6.     """子进程要执行的代码"""
  7.     while True:
  8.         print("---2---")
  9.         time.sleep(1)
  10.  
  11.  
  12. if __name__ == '__main__':
  13.     p = Process(target=run_proc)
  14.     p.start()
  15.     while True:
  16.         print("---1---")
  17.         time.sleep(1)

2.2 进程pid

  1. from multiprocessing import Process
  2. import os
  3. import time
  4.  
  5.  
  6. def run_proc():
  7.     """子进程要执行的代码"""
  8.     print("子进程运行中，pid=%d..." % os.getpid())
  9.     print("子进程将要结束...")
  10.  
  11.  
  12. if __name__ == '__main__':
  13.     print("父进程pid:%d" % os.getpid())
  14.     p = Process(target=run_proc)
  15.     p.start()
'
运行

2.3 给子进程指定的函数传递参数

  1. from multiprocessing import Process
  2. import os
  3. from time import sleep
  4.  
  5.  
  6. def run_proc(name, age, **kwargs):
  7.     for i in range(10):
  8.         print("子进程运行中，name=%s, age=%d, pid=%d..." % (name, age, os.getpid()))
  9.         print(kwargs)
  10.         sleep(0.2)
  11.  
  12.  
  13. if __name__ == '__main__':
  14.     p = Process(target=run_proc, args=('test', 18), kwargs={"m": 20})
  15.     p.start()
  16.     sleep(1)
  17.     p.terminate()
  18.     p.join()
'
运行

2.4 进程间不同享全局变量

  1. from multiprocessing import Process
  2. import os
  3. import time
  4.  
  5.  
  6. nums = [11, 22]
  7.  
  8.  
  9. def work1():
  10.     """子进程要执行的代码"""
  11.     print("in process1 pid=%d, num=%s" % (os.getpid(), nums))
  12.     for i in range(3):
  13.         nums.append(i)
  14.         time.sleep(1)
  15.         print("in process1 pid=%d, nums=%s" % (os.getpid(), nums))
  16.  
  17.  
  18. def work2():
  19.     """子进程要执行的代码"""
  20.     print("in process2 pid=%d, nums=%s" % (os.getpid(), nums))
  21.  
  22.  
  23. if __name__ == '__main__':
  24.     p1 = Process(target=work1)
  25.     p1.start()
  26.     p1.join()
  27.  
  28.     p2 = Process(target=work2)
  29.     p2.start()
'
运行

3、 进程间同步-Queue

Process之间有时需要通信，操作系统提供了很多机制来实现进程间通信

3.1 Queue类语法说明

q=Queue()  初始化Queue对象，若括号中没有指定最大可接收的消息数量，或数量为负值，代表可接受的消息数量没有上限（直到内存的尽头）

Queue.qsize()  返回当前队列包含的消息数量

Queue.empty()  如果队列为空，返回True,反之False

Queue.full()  如果队列满了，则返回True,反之False

Queue.get([block[,timeout]])  获取队列中的一条消息，然后将其从队列中移除，block默认值为True。1.如果block使用默认值，且没有设置timeout（单位秒）消息队列如果为空，此时程序将被阻塞（停在读取状态），直到从消息队列读取到消息为止，如果设置了timeout，则会等待timeout秒，若还没有读取到任何消息，则抛出Queue.Empty异常2.如果block值为False,消息队列如果为空，则会立刻抛出Queue.Empty异常

Queue.get_nowait()  相当于Queue.get(False)

Queue.put(item,[block[,timeout]])  将item消息写入队列，block默认值为True。1.如果block使用默认值，且没有设置timeout,消息队列如果已经没有空间写入，此时程序将被阻塞（停留在写入状态）直到从消息队列腾出空间为止，如果设置了timeout则会等待timeout秒，若还没空间，则抛出Queue.Full异常 2.如果block值为False,消息队列如果没有空间写入，则会立刻抛出异常

Queue.put_nowait(item)  相当于Queue.put(item,False)

3.2 Queue的使用

可以使用multprocessing模块的Queue实现多进程之间数据传递，Queue本身时一个消息队列程序。

  1. from multiprocessing import Queue
  2.  
  3.  
  4. q = Queue(3)  # 初识一个Queue对象，最多可接收三条put消息
  5. q.put("消息1")
  6. q.put("消息2")
  7. print(q.full())
  8. q.put("消息3")
  9. print(q.full())
  10.  
  11.  
  12. try:
  13.     q.put("消息4", True, 2)
  14. except:
  15.     print("消息队列已满，现在消息数量:%s" % q.qsize())
  16.  
  17. try:
  18.     q.put_nowait("消息4")
  19. except:
  20.     print("消息队列已满，现在消息数量:%s"% q.qsize())
  21.  
  22. # 推荐的方式，先判断消息队列是否已满，再写入
  23. if not q.full():
  24.     q.put_nowait("消息4")
  25.  
  26. # 读取消息时，先判断消息队列是否为空，再读取
  27. if not q.empty():
  28.     for i in range(q.qsize()):
  29.         print(q.get_nowait())
'
运行

3.3 Queue实例

  1. from multiprocessing import Process, Queue
  2. import os, time, random
  3.  
  4.  
  5. # 写数据进程执行的代码
  6. def write(q):
  7.     for value in ['A', 'B', 'C']:
  8.         print("Put %s to queue..." % value)
  9.         q.put(value)
  10.         time.sleep(random.random())
  11.  
  12. # 读数据进程执行的代码
  13. def read(q):
  14.     while True:
  15.         if not q.empty():
  16.             value = q.get(True)
  17.             print("Get %s form queue..." % value)
  18.             time.sleep(random.random())
  19.         else:
  20.             break
  21.  
  22.  
  23. if __name__ == '__main__':
  24.     # 父进程创建Queue,并传给各个子进程
  25.     q = Queue()
  26.     pw = Process(target=write, args=(q,))
  27.     pr = Process(target=read, args=(q,))
  28.     # 启动子进程pw，写入
  29.     pw.start()
  30.     # 等待pw结束
  31.     pw.join()
  32.     # 启动子进程pr,读取
  33.     pr.start()
  34.     # pr进程里是死循环，无法等待其结束，只能强行终止
  35.     pr.join()
  36.     print('')
  37.     print("所有数据都写入并且读完")
'
运行

4、进程间同步-Lock

锁是为了保证数据一致性，比如读写锁，每个进程给一个变量增加1，但是如果在一个进程读取但是还没有写入的时候，另外的进程也同时读取了，并写入该值，则最后写入的值是错误的，这个时候就需要加锁来保持数据的一致性。

通过使用Lock来控制一段代码在同一时间只能被一个进程执行。Lock对象的两个方法，acquire()用来获取锁，release()用来释放锁。当一个进程调用acquire()时，如果锁的状态为unlocked并返回，这时该进程即获得了锁，如果锁的状态为locked，那么调用acquire（）的进程则阻塞。

1.Lock的语法说明

lock = multiprocessing.Lock():创建一个锁

lock.acquire():获取锁

lock.release():释放锁

with lock:自动获取、释放锁，类似于with open() as f

2.程序不加锁时：

  1. import multiprocessing
  2. import time
  3.  
  4.  
  5. def add(num, value):
  6.     print("add{0}:num={1}".format(value, num))
  7.     for i in range(0, 2):
  8.         num += value
  9.         print("add{0}:num={1}".format(value, num))
  10.         time.sleep(1)
  11.         
  12.  
  13. if __name__ == '__main__':
  14.     lock = multiprocessing.Lock()
  15.     num = 0
  16.     p1 = multiprocessing.Process(target=add, args=(num, 1))
  17.     p2 = multiprocessing.Process(target=add, args=(num, 2))
  18.     p1.start()
  19.     p2.start()
'
运行

3.程序加锁时

  1. import multiprocessing
  2. import time
  3.  
  4.  
  5. def add(num, value, lock):
  6.     try:
  7.         lock.acquire()
  8.         print("add{0}:num{1}".format(value, num))
  9.         for i in range(0, 2):
  10.             num += value
  11.             print("add{0}:num={1}".format(value, num))
  12.             time.sleep(1)
  13.     except Exception as err:
  14.         raise err
  15.     finally:
  16.         lock.release()
  17.  
  18.  
  19. if __name__ == '__main__':
  20.     lock = multiprocessing.Lock()
  21.     num = 0
  22.     p1 = multiprocessing.Process(target=add, args=(num, 1, lock))
  23.     p2 = multiprocessing.Process(target=add, args=(num, 2, lock))
  24.     p1.start()
  25.     p2.start()
'
运行

只有当其中一个进程执行完成后，其他的进程才会执行，且谁先抢到锁谁先执行

5、进程池Pool

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

multiprocessing.pool.Pool([processes[,initializer[,initargs[,maxtasksperchild[,context]]]]])

参数说明：

processes:工作进程数目，如果processes为None,则使用os.cpu_count()返回的值

initializer:如果initializer不为None,则每个工作进程将会在启动时调用initializer(*initargs)

maxtasksperchild:一个工作进程在它退出或者被一个新的工作进程代替之前能完成的任务数量，为了释放未使用的资源

context:用于指定启动的工作进程的上下文

两种方式向进程池提交任务：

apply(func[,args[,kwds]]):阻塞方式

apply_async(func[args[,kwds]]):非阻塞方式。使用非阻塞方式调用函数func(并行执行，堵塞方式必须等待上一个进程退出才能执行下一个进程，args传递给func的参数列表，kwds为传递给func的关键字参数列表)

multiprocessing.Pool常用函数：

close()  关闭Pool，使其不再接受新的任务

terminate()  不管任务是否完成，立即终止

join()  主进程阻塞，等待子进程退出，必须在close或terminate之后使用

5.2 Pool实例

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

  1. from multiprocessing import Pool
  2. import os, time, random
  3.  
  4.  
  5. def worker(msg):
  6.     t_start = time.time()
  7.     print("%s开始执行，进程号为%d" % (msg, os.getpid()))
  8.     # 生成0-1之间浮点数
  9.     time.sleep(random.random()*2)
  10.     t_stop = time.time()
  11.     print(msg, "执行完毕，耗时%0.2f" % (t_stop-t_start))
  12.  
  13.  
  14. if __name__ == '__main__':
  15.     po = Pool(3)  # 定义一个进程池，最大进程数为3
  16.     for i in range(0, 10):
  17.         # Pool().apply_async(要调用的目标,(传递给目标的参数元组,))
  18.         # 每次循环将会用空闲下来的子进程取调用目标
  19.         po.apply_async(worker, (i,))
  20.  
  21.     print("-----start-----")
  22.     po.close()  # 关闭进程池，关闭后po不再接收新的请求
  23.     po.join()  # 等待po中所有子进程执行完成，必须放在close语句之后
  24.     print("-----stop-----")
  25.  
  26. """
  27. from multiprocessing import Pool
  28. import os, time, random
  31. def worker(a):
  32.     t_start = time.time
  33.     print('%s开始执行,进程号为%d' % (a, os.getpid))
  34.     time.sleep(random.random() * 2)
  35.     t_stop = time.time
  36.     print(a, "执行完成,耗时%0.2f" % (t_stop - t_start))
  39. if __name__ == '__main__':
  40.     po = Pool(3)  # 定义一个进程池
  41.     for i in range(0, 10):
  42.         po.apply_async(worker, (i,))  # 向进程池中添加worker的任务print("--start--")
  43.         po.close()
  44.         po.join()
  45.         print("--end--")
  46. """
'
运行

5.3 进程池中的Queue

如果要使用Pool创建进程，就需要使用multiprocessing.Manager()中Queue()而不是multiprocessing.Queue(),否则会得到一台错误信息：RuntimeError:Queue object should only be shared between processes whrough inheritance

  1. from multiprocessing import Manager, Pool
  2. import os,time,random
  3.  
  4.  
  5. def reader(q):
  6.     print("reader启动(%s),父进程为(%s)" % (os.getpid(), os.getpid()))
  7.     for i in range(q.qsize()):
  8.         print("reader从Queue获取到消息:%s" % q.get(True))
  9.  
  10. def writer(q):
  11.     print("writer启动(%s),父进程(%s)" % (os.getpid(), os.getpid()))
  12.     for i in "itcast":
  13.         q.put(i)
  14.  
  15.  
  16. if __name__ == "__main__":
  17.     print("(%s)start" % os.getpid())
  18.     q = Manager().Queue()
  19.     po = Pool()
  20.     po.apply_async(writer, (q,))
  21.  
  22.     time.sleep(1)
  23.  
  24.     po.apply_async(reader, (q,))
  25.     po.close()
  26.     po.join()
  27.     print("(%s) End" % os.getpid())
'
运行

6、进程、线程对比

6.1功能

进程：能够完成多任务，比如在一条电脑上能够同时运行多个QQ

线程：能够完成多任务，比如一个QQ中多个聊天窗口

定义的不同：线程是系统进行资源分配和调度的一个独立单位

线程是进程的一个实体，是CPU调度和分配的基本单位，它是比进程更小的能独立运行的基本单位，线程自己基本上不拥有系统资源，只拥有一点在运行中必不可少的资源（如程序计数器，一组寄存器和栈）但是它可与同一个进程的其他线程共享进程所拥有的全部资源

6.2 区别

一个程序至少有一个进程，一个进程至少有一个线程

线程的划分尺度小于进程（资源比进程少）使得多线程程序的并发性高

进程在执行过程中拥有独立的内存单元，而多个线程共享内存，从而极大地提高了程序的运行效率

线程不能独立执行，必须依存在进程中

可以将进程理解为工厂中的一条流水线，而其中的线程就是流水线中的工人

6.3 优缺点

线程：线程执行开销小，但不利于资源的管理和保护

进程：进程执行开销大，但利于资源的管理和保护

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/我家自动化/article/detail/860661
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号