python中进程间通信及设置状态量控制另一个进程

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

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