Python 中多进程是由 multiprocessing 模块提供的
import time
from multiprocessing import Process
def test(i):
print("子进程开始")
print(i)
time.sleep(2)
print("子进程结束")
if __name__ == "__main__":
p_ls = []
for i in range(3):
p = Process(target=test, args=(i, ))
p.start()
p_ls.append(p)
for i in range(3):
p_ls[i].join(()
还有一种方法，使用类：
from multiprocessing import Process
class MyProcess(Process):
def __init__(self):
super().__init__()
def run():
pass
p = MyProcess()
p.start()
p.join()
终止一个进程：p.terminate()
判断一个进程是否存活： p.is_alive
查看一个进程的 id：p.pid
查看进程对象名称：p.name
守护进程会随着主进程的代码执行完毕而结束
在 p.start() 之前设置 p.daemon = True
加锁。涉及到数据改动的时候，目的是保证数据安全。
from multiprocessing import Lock
lock = Lock()
lock.acquire()
lock.release()
进程同步控制 ---- 锁，信号量，事件
- 信号量 相当于多个锁，可以规定获取锁个数的大小。其内部相当于加了一个计数器，acquire() 的时候计数器 -1，release() +1
from multiprocessing import Process, Semaphore
def test(i, sem):
sem.acquire()
pass
sem.release()
if __name__ =="__main__":
sem = Semaphore(4)
for i in range(4):
p = Process(target=test, args=(i, sem,))
p.start()
- 事件
一个信号可以使所有的进程都进入阻塞状态，也可以解除
创建之初默认是阻塞的
from multiprocessing import Event
e = Event()
e.is_set() --> False
e.set() --> 将事件的状态设置为 True
e.wait() --> 依据 is_set 的值来决定是否阻塞，相当于监视 is_set 的状态，然后根据其状态来做某些事情
e.clear() --> 将事件的状态设置为 False 阻塞
例如一个简单的红绿灯事件
import time
import random
from multiprocessing import Process, Event
def car(e, i):
if not e.is_set():
print('car %s 在等待'%(i) )
e.wait() --> 阻塞，直到得到事件状态变为 True
print('car %s 通过'%i)
def light(e):
while True:
if e.is_set(): --> 开始时是 False，所以绿灯亮
e.clear()
print("红灯亮了")
else:
e.set() --> 设置为 True，红灯亮
print("绿灯亮了")
time.sleep(2)
if __name__ == "__main__":
e = Event()
traffic = Process(target=light, args=(e, ))
traffic.start()
for i in range(10):
c = Process(target=cars, args=(e, i))
c.start()
time.sleep(random.randint())
进程间通信 IPC ----- 队列和管道
- 队列
# 先进先出
from multiprocessing import Queue
q = Queue(5)
q.put(1)
q.get()
q.full() -->> 队列是否满了
q.empty() -->> 判断队列是否空
q.get_nowait() ->> 如果队列空了，那么它会报错
消费者与生产者
import time
import random
from multiprocessing import Process, Queue
def producer(q):
for i in range(10)：
time.sleep(random.randint(1, 2))
food = "产品%s"%i
print('生产者生产了一个东西')
q.put(food)
def consumer(q):
while True:
food = q.get()
if food is None:
break
print(food)
time.sleep(random.random())
if __name__ == "__main__":
q = Queue()
p = Process(target=produce, args=(q,))
p.start()
c = Process(target=consume, args=(q, ))
c.start()
p.join()
q.put(None)
但是上面的代码有一个问题，就是 q.put(None) 的频繁，有几个消费者，那么就需要向队列中 put 几个值，这种情况可以使用 JoinableQueue
from multiprocessing import JoinableQueue
def producer(q):
xxxx
q.join() -->> 阻塞，直到队列中的所有数据全被处理完毕
def consumer(q):
xxxx
q.task_done()
if __name__ == "__main__":
q = JoinableQueue()
c.daemon = True
- 管道
from multiprocessing import Pipe
conn1, conn2 = Pipe()
conn1.send('12345')
conn2.recv()
进程池
主要是为了效率问题，开启单个进程会耗费很多资源。进程池就是先创建好多个进程，然后从进程池中取到进程，提高操作系统调度进程的利用率。一般 CPU 个数 + 1 是最大开启进程的数量。
from multiprocessing import Pool
def func(n):
for i in range(10):
n += 1
print(n)
if __name__ == "__main__":
pool = Pool(5)
pool.map(func, <可迭代对象>range(100))
# map 是自带 close 和 join 的，最后一次返回结果
res_lst = []
# 还有一个 apply 方法
for i in range(20):
p.apply(func, args=(i, )) --> 同步提交的方式
res = p.apply_sync(func, args=(i, )) --> 异步提交方式
# res.get() --> 阻塞等待结果
# 如果需要不阻塞，那么可以这样做
res_lst.append(res)
# 最后可以得到结果
for res in res_lst:
print(res.get())
# 与 apply_async 配合使用
p.close() --> 结束进程池接受任务
p.join() --> 感知进程池中的任务结束
进程池实现 socket 服务端
import socket
from multiprocessing import Pool
def func(conn):
conn.send(b'hello')
msg = conn.recv(1024).decode('utf-8')
print(msg)
con.close()
if __name__ == "__main__"：
p = Pool(5)
sk = socket.socket()
sk,bind(('127.0.0.1', 8000))
sk.listen()
while True:
conn, addr = sk.accept()
p.apply_async(func, args=(conn, ))
sk.close()
另外进程池还有回掉函数，意思就是在 任务执行完毕后执行指定的回调函数，任务返回的数据作为回调函数的参数。回调函数的参数来源只有进程返回的数据。回调函数是在主进程中执行的
def func1():
pass
return x
def func2(x):
pass
return
if __name__ == "__main__":
pool = Pool(5)
p.apply_async(func, args=(), callback=func2)
回调函数经常用于爬虫，因为网络延迟和下载很耗时。
例如
import requests
from multiprocessing import Pool
def get(url):
res = requests.get(url)
if res.status_code == 200:
return url, res.content
def call_back(args):
url, content = args
print(url, len(content))
if __name__ == "__main__":
url_lst = []
p = Pool(5)
for url in url_lst:
p.apply_async(get, args=(url,), callback=call_back)
p.close()
p.join()
