python中进程间通信及设置状态量控制另一个进程
作者:PT、小小马 发布时间:2022-01-28 08:02:57
一、python中进程间通信
业务场景:在当前遇到的业务场景中,我们需要启一个间隔任务,这个间隔任务跑一个算法,然后把算法的结果进行一些处理,并入库。任务目前间隔是一小时,算法运行时间要50多分钟,留给结果处理的时间并不多,所以有可能会出现超时。目前来说,优化方向在算法上会更为合理,因为结果处理本来就不用很多时间。但是在这个业务场景下,想要把结果处理的时间进行无限压缩,压缩到0,其实也是可以实现的,说是压缩为0,实际上就是在算法执行完成后,再启一个进程去处理,这样就不会由于需要进行数据处理而影响到算法的运行,将算法和结果处理分为两个独立的进程去处理。在最开始的程序中,是把算法运行和结果处理作为一个周期,而现在是把算法运行和结果处理分为两个周期去处理。
技术实现方案:
启动二个进程,其中一个运行算法,在算法运行结束后,发送一个状态值到另外一个进程,另外一个进程在收到状态量后启动数据处理即可。两个进程间互不影响即可。其实也相当于算法进程控制数据处理进程
测试场景构造代码:
from multiprocessing import Process,Pipe
import time
import sys
import os
def send_message(conn):
for i in range(1000):
print('send_message:%d'%i)
print(os.getpid())
conn.send(i)
time.sleep(3)
def send_message1(conn):
# for i in range(1000):
print(conn.recv())
while True:
if conn.recv() % 5 == 0:
print(' today is nice day')
time.sleep(1)
if __name__ == '__main__':
#创建一个进程通信管道
left,right = Pipe()
t1 = Process(target=send_message,args=(left,))
t2 = Process(target=send_message1,args=(right,))
t1.start()
t2.start()
在这个案例场景下有一些需要注意的点:
一、time.sleep()的问题,睡眠指定时间,总是会出错,具体的出错原因到现在也没有找到,这是原来出现的问题,在这里没有做长时间的测试,所以不一定会出现,但是还是要注意
二、代码实现中与上述的描述差异有一些,如未启用调度任务,只是启了一个间隔运行的任务。
三、数据处理进程一直处理空跑状态,会造成资源的浪费(更合理的应该是形成阻塞状态,但是对于阻塞状态的构造缺乏认知,所以先牺牲资源
四、在上述描述的需求中,在算法运行及数据处理的上一节点还有一个调度任务在控制,这里未做出体现,其实应该把定时任务和数据处理作为两个周期独立出来才更符合上述描述中的需求。
二、设置状态量控制另一个进程
业务场景:在当前遇到的业务场景中,我们需要启一个间隔任务,这个间隔任务跑一个算法,然后把算法的结果进行一些处理,并入库。任务目前间隔是一小时,算法运行时间要50多分钟,留给结果处理的时间并不多,所以有可能会出现超时。目前来说,优化方向在算法上会更为合理,因为结果处理本来就不用很多时间。但是在这个业务场景下,想要把结果处理的时间进行无限压缩,压缩到0,其实也是可以实现的,说是压缩为0,实际上就是在算法执行完成后,再启一个进程去处理,这样就不会由于需要进行数据处理而影响到算法的运行,将算法和结果处理分为两个独立的进程去处理。在最开始的程序中,是把算法运行和结果处理作为一个周期,而现在是把算法运行和结果处理分为两个周期去处理。
上面的解决方案中只涉及到了启用两个进程去运行两个任务,并未涉及到启用定时任务框架,所以可能会显得和上述的业务场景不一致,所以在这里重新解决一下。上面也是没有问题的,只是把定时任务框架也作为一个任务去处理即可。然后在定时任务运行完程后,向另外一个进程传入一个参数,作为启动另一个进程的状态量即可。当然,在这里,两个进程还是完全占满的,即处理阻塞状态。对于资源的利用还是没有完全达到最好。后续再考虑使用进程池的方式,看是否可以让其中的一个进程运行完后直接释放资源。
技术解决方案如下:
from multiprocessing import Process,Pipe
import time
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.schedulers.blocking import BlockingScheduler
from apscheduler.schedulers.asyncio import AsyncIOScheduler
# schedule = BackgroundScheduler()
schedule = BlockingScheduler(timezone="Asia/Shanghai")
# schedule = AsyncIOScheduler(timezone="Asia/Shanghai")
def algorithm(conn):
print('start_run')
conn.send('please run')
# time.sleep(5)
def worth_result(conn):
while True:
if conn.recv() == 'please run':
print(conn.recv() + ' very nice!')
def time_job(conns):
schedule.add_job(func=algorithm,trigger='interval',seconds=5,args=(conns,))
schedule.start()
if __name__ == '__main__':
left,right = Pipe()
t1 = Process(target=time_job,args=(left,))
t2 = Process(target=worth_result,args=(right,))
t1.start()
t2.start()
在这里还有一些点需要说明,定时任务选择那一种类型其实都没有关系,阻塞和非阻塞其实没有关系,因为我们在这里是直接启了两个进程,每个进程间是相互独立的,并非是在定时任务下启用的两个进程,所以不会影响的。
关于这个解决方案还有的问题:
一、上述所说,两个进程是占满的,所以对于资源来说,两个进程的利用率一直很高
二、扩展性不足,如果在这个程序中还有其他需要处理的过程,就需要再添加进程,或者把他添加到当前的进程之下,代码重构会比较麻烦一些
三、整个任务的控制不足,需要加以完善。比如对于运行状态一些控制及查看,一般程序如果运行时间较长的话,我们应该添加这样的接口,否则启动后如果没有出结果,我们是不知道其运行状态,有一点被动
四、关于三,使用logging库,应该是可以直接去输出其日志,但是日志库作为第三方库,相当于是对整个运行状态进行监控,会不会再占用一个进程,这个需要去测试
五、完备性及容灾处理,如果程序由于资源等其他问题挂掉后,会有一些数据冗余下来,也就是一些算法未进行处理,这个时候需要考虑怎么样去补数据?原始文件如果没有保留下来呢?而且如果这些数据是极重要的数据该怎么处理?如果程序挂掉后,应该如何快速的去处理呢?直接重启吗?
六、如果数据处理的进程所用的时间比算法还多,那该怎么办?目前的业务来看,是远低于的,但是如果是远高于呢?可否将处理工作进行分配,利用多台机器来处理,然后再把结果合并起来?
分布式处理的思想越来越浓。
来源:https://blog.csdn.net/qq_44862918/article/details/124842372
猜你喜欢
- config.php<?phpdefine('APP_DIR', dirname(__FILE__));define(
- 创建项目和应用django-admin startproject zqxt_views(项目名)cd zqxt_viewspython ma
- 创建py文件总是为txt格式问题记录写代码过程中创建.py文件时,一直正常,但创建名称为train.py文件时总是为txt格式,即使选择了p
- Conda GuideConda简介conda是一个包,依赖和环境管理工具,适用于多种语言,如: Python, R, Scala, Jav
- 本文实例讲述了php购物车实现方法。分享给大家供大家参考。具体分析如下:这里我们为你提供个简单的php购物车代码,从增加购物产品与发生购买了
- 很多的网站都有记数器,用来记录网站的访问量,这给网站管理员即时了解本网站的运行及访问情况提供了很多的方便。笔者研究过很多用ASP编写的计数器
- 一、基础概述机器学习(Machine Learing)是一门多领域交叉学科,涉及概率论、统计学、逼近论、凸分析、算法复杂度理论等多门学科。机
- The WeekdayName function returns the weekday name of a specified day o
- 本文实例为大家分享了python视频按帧截取图片工具的具体代码,供大家参考,具体内容如下描述:将一个视频流按帧数截取大量的图片用途:AI的数
- pytorch 输出中间层特征:tensorflow输出中间特征,2种方式:1. 保存全部模型(包括结构)时,需要之前先add_to_col
- 前言本文旨在记录使用Flask框架过程中与前端Vue对接过程中,存在WebSocket总是连接失败导致前端取不到数据的问题。以及在使用Web
- 1. 面向对象编程 OOP ( Object Oriented Programming) 即面向对象编程。面向对象编程是一种&nb
- 遇到一个难题,在无物理键盘情况下,通过页面软键盘在页面文本框输入汉字,不知道51js的各位大牛有没有遇到过这种需求,如果遇到过是如何解决的,
- 在asp中利用excel的一个方法是将excel文件作为一个数据库进行链接,然后的操作和对access数据库操作类似.但是这个方法不是总能有
- 本文实例讲述了Python图像处理之简单画板实现方法。分享给大家供大家参考,具体如下:Python图像处理也是依赖opencv的Python
- 在进行matplotlib画图的时候,经常会出现这个的报错,虽然知道是因为没有对应的字体的原因,但是,将字体下载后放到目标路径下,仍然没有办
- 在一个排列中,如果一对数的前后位置与大小顺序相反,即前面的数大于后面的数,那么它们就称为一个逆序。一个排列中逆序的总数就称为这个排列的逆序数
- 在想把CIFAR-10 的16进制数据转为图像的时候需要用到imsave(),但是在import scipy.misc 之后显示该库内没有i
- 今天我们整理了ip地址和身份证的javascript验证方法。虽然ip地址和身份证的验证不是很经常会遇到,但是大家也可以研究一下js代码,里
- 去除HTML代码中所有标签<% '****************************** '函数:RemoveH