Python进程间通信实战：利用Pipe与状态量解耦算法及数据处理

一、Python中进程间通信

在当前的业务场景中，我们需要启动一个周期性任务来运行特定算法，随后对算法结果进行预处理并写入数据库。目前任务调度间隔为一小时，而算法执行耗时超过50分钟，留给结果处理的时间窗口非常狭窄，极易导致处理超时。虽然从长远看，优化算法本身的性能更为合理（因为结果处理本身耗时较短），但在当前业务约束下，若希望将结果处理的时间压缩至极限甚至接近零，也是可行的。具体做法是：在算法执行完毕后，启动另一个独立的进程来处理数据。这样，数据处理不再占用算法运行的资源，两者被划分为两个完全独立的进程。原有的程序逻辑是将算法运行和结果处理合并为一个执行周期，而新的方案则是将这两个阶段拆分为两个独立的周期进行调度。

技术实现方案：

核心思路是启动两个独立的进程。其中一个进程专门负责运行算法，当算法执行结束时，通过通信机制发送一个状态值给另一个进程。接收进程在检测到该状态量后，立即启动数据处理流程。两个进程之间相互隔离、互不干扰，实质上构成了由算法进程对数据处理进程的控制关系。

测试场景构造代码：

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多分钟，留给结果处理的时间很少，可能导致超时。目前看来，优化算法本身更为合理，因为结果处理耗时本就不长。但在当前业务限制下，若希望将结果处理时间无限压缩至零，也是可行的。所谓“压缩为零”，实际上是指在算法执行完成后，再启动一个进程去处理数据。这样，数据处理不会占用算法运行的时间，两者被分为两个独立的进程。最初的程序是将算法运行和结果处理作为一个周期，而现在将其分为两个周期处理。

上述解决方案仅涉及启动两个进程运行两个任务，并未引入定时任务框架，因此可能与前述业务场景描述存在不一致。为此，这里重新提供一种解决方案。之前的方案依然有效，只需将定时任务框架也作为一个任务进程来处理即可。在定时任务运行结束后，向另一个进程传入一个参数作为启动状态量。需要注意的是，此时两个进程依然处于完全占用状态，即处理阻塞状态，资源利用率并未达到最优。后续可以考虑使用进程池（Process Pool）的方式，尝试在其中一个进程完成后直接释放资源，以优化资源利用。

技术解决方案如下：

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库可以直接输出运行日志。但日志库作为第三方库，是否会对整个运行状态监控造成额外进程开销，需要进一步测试验证。
• 第五，完备性及容灾处理。如果程序因资源不足等原因挂掉，会导致部分数据冗余（即算法已运行但未处理的数据）。此时需要考虑如何补录数据？如果原始文件未保留，且这些数据至关重要，该如何处理？程序挂掉后，应如何快速恢复？是直接重启吗？这些都需要在容灾设计中予以考虑。
• 第六，如果数据处理进程所用的时间比算法还多，该怎么办？目前的业务场景中，处理时间远低于算法时间，但如果未来处理时间远高于算法时间，该如何应对？是否可以将处理工作进行拆分，利用多台机器并行处理，然后再合并结果？

分布式处理的思想在此场景下显得愈发重要。

至此，关于Python中进程间通信及利用状态量控制另一个进程的详细介绍到此结束。更多相关Python进程通信内容，请搜索本站以前的文章或继续浏览下面的相关文章，希望大家以后多多支持本站！