一、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库，应该是可以直接去输出其日志，但是日志库作为第三方库，相当于是对整个运行状态进行监控，会不会再占用一个进程，这个需要去测试
• 五、完备性及容灾处理，如果程序由于资源等其他问题挂掉后，会有一些数据冗余下来，也就是一些算法未进行处理，这个时候需要考虑怎么样去补数据？原始文件如果没有保留下来呢？而且如果这些数据是极重要的数据该怎么处理？如果程序挂掉后，应该如何快速的去处理呢？直接重启吗？
• 六、如果数据处理的进程所用的时间比算法还多，那该怎么办？目前的业务来看，是远低于的，但是如果是远高于呢？可否将处理工作进行分配，利用多台机器来处理，然后再把结果合并起来？

分布式处理的思想越来越浓。

免责声明：本站发布的内容（图片、视频和文字）以原创、转载和分享为主，文章观点不代表本网站立场，如果涉及侵权请联系站长邮箱：mmqy2019@163.com进行举报，并提供相关证据，查实之后，将立刻删除涉嫌侵权内容。

猜你喜欢

• 怎样用Python实现烟花秀效果，要用到哪些模块

  这篇文章主要介绍了利用Python写一场新年烟花秀，可以送给朋友还可以当作练习，Python 模块包括：tkinter、PIL、time、random、math，下面来看一下代码实现,需要的朋友可以参考一下

• 如何用Python实现钉钉群应答机器人，完整思路方法是什么

  在聊天工具大肆侵入我们生活各个方面的今天,各种消息无时无刻不在侵扰我们的每一寸时间,这种情况下,一个聊天的机器人就很有必要了.今天,我们来学习一下使用 Python 开发一个钉钉的应答机器人,助你「人生苦短，少回消息」,需要的朋友可以参考下

• 如何Python实现人脸微笑检测的功能，具体过程是什么

  Python可以从图像或视频中检测和识别你的脸，人脸检测与识别也是计算机视觉领域的研究热点之一。本文主要为大家介绍通过Python实现人脸识别之微信检测，需要的同学可以参考一下

• Python编程模块常用的包括什么，怎样实现

  今天就给大家分享一些python常用的编程模块吧，包括文件流的读写及如何删除str中的特定字符，感兴趣的朋友跟随一起学习下吧

• django使用MQTT的方法是什么？如何安装MQTT？

  我们知道Django是由Python写成的，一个开放源代码的Web应用框架。本文主要给大家介绍django使用MQTT的方法，那么什么是MQTT？MQTT也叫做消息队列遥测传输协议，其最大优点在于，可以以极少的代码和有限的带宽，为连接远程设备提供实时可靠的消息服务。简单了解完MQTT，下面我们一起来看看在django中怎么使用MQTT。