linux创建进程
1. Linux创建子进程的原理:
#1). 父进程和子进程, 如果父进程结束, 子进程也随之结束;
#2). x先有父进程, 再有子进程, 通过fork函数实现;
2. fork函数的返回值:调用该方法一次, 返回两次;
#- 产生的子进程返回一个0
#- 父进程返回子进程的pid;
3. Window也能使用fork函数么?
#Windows没有fork函数, Mac有fork函数(Unix -> Linux, Unix-> Mac),
#封装了一个模块multiprocessing
4. 常用方法:
#- os.fork()
#- os.getpid(): 获取当前进程的pid;
#- os.getppid(): parent process id, 获取当前进程的父进程的id号;
import multiprocessing
import os
import time
print('当前进程(pid=%d)正在运行...' %(os.getpid()))
print('当前进程的父进程(pid=%d)正在运行...' %(os.getppid()))
print('正在创建子进程...')
pid=os.fork()
pid2=os.fork()
print('第一个:',pid)
print('第二个:',pid2)
if pid==0:
print('这是创建的子进程,子进程的id为%s,父进程id为%s' %(os.getpid(),os.getppid()))
else:
print('当前父进程[%s]的返回值是%s' %(os.getpid(),pid))
time.sleep(10)
创建子进程_通过实例化process对象创建多进程
import multiprocessing
import threading
import time
def job():
print('当前子进程的名称为%s' %(multiprocessing.current_process()))
#创建一个进程对象
p1=multiprocessing.Process(target=job)
#运行多进程,执行任务
p1.start()
#创建一个进程对象
p2=multiprocessing.Process(target=job)
p2.start()
#等待所有的子进程执行结束,在执行主进程内容
p1.join()
p2.join()
print('任务执行结束...')
派生多进程,通过继承的方式创建子进程
import multiprocessing
import time
class JobProcess(multiprocessing.Process):
#重写Process的构造方法,获取新的属性
def __init__(self,queue):
super(JobProcess, self).__init__()
self.queue=queue
#重写run方法,将执行的任务放在里面即可
def run(self):
print('当前子进程名称为%s' %(multiprocessing.current_process()))
processes=[]
#启动10个进程,来执行需要执行的任务
for i in range(10):
#实例化对象
p=JobProcess(queue=3)
processes.append(p)
#启动多进程,处理需要执行的任务
p.start()
#等待所有的子进程执行结束,在继续执行主进程
[process.join() for process in processes]
time.sleep(3)
#执行主进程
print('任务执行结束...')
mytimeit
import time
def timeit(f):
def wrapper(*args, **kwargs):
start_time = time.time()
res = f(*args, **kwargs)
end_time = time.time()
print("%s运行时间为%s" %(f.__name__, end_time-start_time))
return res
return wrapper
多进程案例
import multiprocessing
import threading
from mytimeit import timeit
class JobProcess(multiprocessing.Process):
def __init__(self,li):
super(JobProcess, self).__init__()
self.li=li
def run(self):
for item in self.li:
sum(item)
class JobThread(threading.Thread):
def __init__(self,li):
super(JobThread, self).__init__()
self.li=li
def run(self):
sum(self.li)
@timeit
def use_multi_process():
# 1). 开启的进程数有瓶颈, 取决于CPU的个数;
# 2). 如果处理的数据比较小, 不建议使用多进程, 因为进程的创建和销毁需要时间,
# 开启的进程数越多,不一定效率越高;;
# 3). 如果处理的数据量足够大, 0<开启的进程数<cpu个数, 开启的进程数越多, 效率越高;
lists = [[1, 2, 3, 4, 5], [2, 3, 4, 5, 6],
[2, 3, 4, 5, 6, 7, 8], [2, 3, 4, 5, 6]] * 10
processes=[]
for li in lists:
p=JobProcess(li)
processes.append(p)
p.start()
# # 对于处理的数据进行分片, 每个进程处理一部分任务;
# for i in range(0, len(lists), 10):
# # print(i, i+1000)
# p = JobProcess(lists[i:i + 10])
# processes.append(p)
# p.start()
#等待所有的子进程执行结束
[process.join() for process in processes]
print('多进程执行结束')
@timeit
def use_multi_thread():
lists = [[1, 2, 3, 4, 5], [2, 3, 4, 5, 6],
[2, 3, 4, 5, 6, 7, 8], [2, 3, 4, 5, 6]] * 10
processes = []
for li in lists:
p = JobThread(li)
processes.append(p)
p.start()
# 等待所有的子进程执行结束
[process.join() for process in processes]
print("多线程执行结束.......")
if __name__=='__main__':
use_multi_process()
use_multi_thread()
守护进程
# 守护线程:
#setDeamon:
#True: 主线程执行结束, 子线程不再继续执行;
#Flase:
import multiprocessing
import time
def job():
name=multiprocessing.current_process()
print('%s开始运行' %(name))
time.sleep(3)
print('%s结束进程' %(name))
if __name__=='__main__':
#启动一个子进程
p1=multiprocessing.Process(target=job,name='use_daemon')
p1.daemon=True
p1.start()
#join等待所有子进程执行结束,在执行主进程
p1.join(3)
#执行主进程
print('%s程序执行结束'%(multiprocessing.current_process()) )
终止进程
import time
import multiprocessing
def job():
print("start.....")
time.sleep(1)
print('end.......')
if __name__ == '__main__':
p = multiprocessing.Process(target=job)
print("Before:", p.is_alive())
p.start() # 启动子进程
print("During:", p.is_alive())
p.terminate() # 终止子进程
print('terminate:', p.is_alive())
p.join() #等待子进程彻底终止
print("joined:", p.is_alive())
进程池之ProcessPoolExecutor
from concurrent.futures import ThreadPoolExecutor
from concurrent.futures import ProcessPoolExecutor
def job(id):
print('start %d' %(id))
print('end %d' % (id))
return id
# 第2种方式: submit
pool = ProcessPoolExecutor(max_workers=4)
# 分配任务给子进程, 并且返回一个Future对象;
f1 = pool.submit(job, 5)
# 获取进程是否执行结束;
f1.done()
# 获取子进程执行的结果
f1.result()
# 第三种方式:map函数
# pool = ProcessPoolExecutor(max_workers=4)
# for res in pool.map(job, range(1,100)):
# print(res)
进程间的通信之生产者和消费者
import multiprocessing
from multiprocessing import Queue
import time
class Producer(multiprocessing.Process):
#往队列写内容
def __init__(self,queue):
super(Producer,self).__init__()
self.queue=queue
def run(self):
for i in range(10):
self.queue.put(i)
time.sleep(2)
print('传递消息,内容为:%s' %(i))
class Consumer(multiprocessing.Process):
def __init__(self,queue):
super(Consumer, self).__init__()
self.queue=queue
def run(self):
# 判断队列是否为空, 如果是, 跳出循环, 不会再去从队列获取数据;
# while not self.queue.empty():
while True:
time.sleep(0.1)
print("读取进程传递的消息:%s" % (self.queue.get()))
if __name__=='__main__':
q=Queue()
p1=Producer(q)
c1=Consumer(q)
p1.start()
c1.start()
p1.join()
c1.terminate()
c1.join()
print('all done')