Python多进程中Win与Java应用同步的互斥/信号量实现问题

需求概述 #

需要实现Windows应用与Java应用的循环同步交互，流程如下：

• 启动Windows应用与Java应用
• 在Java应用中执行命令，等待响应
• 将Java响应作为输入传入Windows应用，执行命令并等待响应
• 将Windows响应作为输入传入Java应用，重复上述循环
• 两个应用需保持活跃进程，在控制台内执行命令

现有问题 #

尝试用互斥锁（mutex）或二元信号量（binary semaphore）实现进程同步与上下文切换时遇到困难，现有代码中标记的TODO部分无法完成。

解决方案 #

使用multiprocessing模块的Semaphore（二元信号量）实现进程交替执行，同时用Queue完成进程间的命令与响应传递（多进程内存不共享，无法直接通过类属性传递数据）。

修改后的完整代码 #

import datetime
import time
import multiprocessing
import subprocess

# 辅助函数（用户可根据实际需求调整实现）
def write_into_file(file_path, line):
    with open(file_path, 'a', encoding='utf-8') as f:
        f.write(line + '\n')

def parse_lines_and_get_command(lines):
    # 示例逻辑：从输出中提取下一个命令，需根据应用实际输出格式修改
    for line in lines:
        if 'next_command:' in line:
            return line.split(':')[-1].strip()
    return ''

class Win_App():
    def __init__(self, java_sem, win_sem, java_queue, win_queue):
        self.java_sem = java_sem  # 控制Java进程执行的信号量
        self.win_sem = win_sem    # 控制Windows进程执行的信号量
        self.java_queue = java_queue  # 向Java进程传递命令的队列
        self.win_queue = win_queue    # 接收Java进程命令的队列
        self.output_log = []

    def execute_win_application(self):
        exe_path = 'path_to_exe'
        init_command = 'command_to_execute'
        # 启动Windows应用控制台进程，开启行缓冲实时读取输出
        win_process = subprocess.Popen(
            'cmd.exe', 
            stdin=subprocess.PIPE, 
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT, 
            text=True,
            bufsize=1
        )
        # 发送初始化命令并刷新输入缓冲区
        win_process.stdin.write(f'{exe_path}\n')
        win_process.stdin.write(f'{init_command}\n')
        win_process.stdin.flush()
        
        print('Windows Application Started!')
        
        while True:
            # 等待Java进程传递命令，阻塞直到信号量被释放
            self.win_sem.acquire()
            
            # 从队列获取输入命令
            input_cmd = self.win_queue.get()
            if input_cmd == 'exit':
                print('Windows Application exiting...')
                win_process.stdin.write('exit\n')
                win_process.stdin.flush()
                break
            
            # 执行命令并刷新缓冲区
            win_process.stdin.write(f'{input_cmd}\n')
            win_process.stdin.flush()
            
            # 读取应用输出（需根据实际输出结束标志调整循环逻辑）
            output_lines = []
            while True:
                line = win_process.stdout.readline()
                if not line:
                    break
                stripped_line = line.strip()
                write_into_file("win_app_output.log", stripped_line)
                output_lines.append(stripped_line)
                self.output_log.append(stripped_line)
            
            # 解析输出得到下一个给Java的命令
            next_java_cmd = parse_lines_and_get_command(output_lines)
            if next_java_cmd == '-1':
                next_java_cmd = 'exit'
            
            # 将命令传入队列，释放信号量唤醒Java进程
            self.java_queue.put(next_java_cmd)
            self.java_sem.release()
        
        # 等待进程终止并清理
        win_process.wait()
        print('Windows Application terminated.')
         
class Java_App():
    def __init__(self, java_sem, win_sem, java_queue, win_queue):
        self.java_sem = java_sem
        self.win_sem = win_sem
        self.java_queue = java_queue
        self.win_queue = win_queue
        self.output_log = []
        
    def execute_java_application(self):
        jav_path = 'path_to_java_app'
        init_command = 'command_to_execute'
        # 启动Java应用控制台进程
        jav_process = subprocess.Popen(
            'cmd.exe', 
            stdin=subprocess.PIPE, 
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT, 
            text=True,
            bufsize=1
        )
        # 发送初始化命令
        jav_process.stdin.write(f'{jav_path}\n')
        jav_process.stdin.write(f'{init_command}\n')
        jav_process.stdin.flush()
        
        print('Java Application Started!')
        
        # 执行初始命令并获取输出
        first_cmd = 'initial_command'  # 替换为实际初始命令
        jav_process.stdin.write(f'{first_cmd}\n')
        jav_process.stdin.flush()
        
        output_lines = []
        while True:
            line = jav_process.stdout.readline()
            if not line:
                break
            stripped_line = line.strip()
            write_into_file("jav_app_output.log", stripped_line)
            output_lines.append(stripped_line)
            self.output_log.append(stripped_line)
        
        # 解析初始输出，获取给Windows的第一个命令
        next_win_cmd = parse_lines_and_get_command(output_lines)
        if next_win_cmd == 'exit':
            jav_process.stdin.write('exit\n')
            jav_process.stdin.flush()
            jav_process.wait()
            print('Java Application terminated.')
            return
        
        # 传递命令并唤醒Windows进程
        self.win_queue.put(next_win_cmd)
        self.win_sem.release()
        
        while True:
            # 等待Windows进程传递命令
            self.java_sem.acquire()
            
            input_cmd = self.java_queue.get()
            if input_cmd == 'exit':
                print('Java Application exiting...')
                jav_process.stdin.write('exit\n')
                jav_process.stdin.flush()
                break
            
            # 执行命令
            jav_process.stdin.write(f'{input_cmd}\n')
            jav_process.stdin.flush()
            
            # 读取输出
            output_lines = []
            while True:
                line = jav_process.stdout.readline()
                if not line:
                    break
                stripped_line = line.strip()
                write_into_file("jav_app_output.log", stripped_line)
                output_lines.append(stripped_line)
                self.output_log.append(stripped_line)
            
            # 解析下一个给Windows的命令
            next_win_cmd = parse_lines_and_get_command(output_lines)
            if next_win_cmd == 'exit':
                next_win_cmd = 'exit'
            
            # 传递命令并唤醒Windows进程
            self.win_queue.put(next_win_cmd)
            self.win_sem.release()
        
        # 清理进程
        jav_process.wait()
        print('Java Application terminated.')    

##### MAIN PROGRAM #####

if __name__ == '__main__':
    # 初始化二元信号量：初始值为0，进程调用acquire时会阻塞，直到被release唤醒
    java_sem = multiprocessing.Semaphore(0)
    win_sem = multiprocessing.Semaphore(0)
    
    # 初始化进程间通信队列：分别存储给Java和Windows的命令
    java_queue = multiprocessing.Queue()
    win_queue = multiprocessing.Queue()
    
    # 初始化应用实例，传入同步对象
    win = Win_App(java_sem, win_sem, java_queue, win_queue)
    jav = Java_App(java_sem, win_sem, java_queue, win_queue)

    # 启动进程
    p1 = multiprocessing.Process(target=win.execute_win_application)
    p2 = multiprocessing.Process(target=jav.execute_java_application)
    
    p1.start()
    print("Windows App started!")
    p2.start()
    
    # 等待子进程全部结束
    p1.join()
    p2.join()
    print('All processes completed.')

关键实现说明 #

1. 进程间通信：使用multiprocessing.Queue传递命令，解决多进程内存不共享的问题，避免直接操作类属性导致的数据无法同步。
2. 二元信号量同步：
   • java_sem和win_sem初始值为0，通过acquire()阻塞当前进程，release()唤醒目标进程，严格控制两个应用的交替执行顺序。
   • Java进程先执行初始命令，完成后将命令传入队列并释放win_sem唤醒Windows进程；Windows进程执行完成后，再传递命令并释放java_sem唤醒Java进程，形成循环。
3. 子进程IO处理：设置bufsize=1开启行缓冲，确保实时读取应用输出；每次发送命令后调用flush()，确保命令被控制台进程接收。
4. 终止逻辑：当解析到exit命令时，终止当前进程并传递终止信号给另一个进程，确保两个应用同步退出。

内容的提问来源于stack exchange，提问作者Raj Sawant