如何在Robot Framework中并行运行多个关键字？多进程/线程尝试失败

核心问题点 #

1. 进程启动逻辑错误：创建Process时直接执行了方法（importABR.abr1_keyword(self)），导致方法在主进程中同步运行，而非交给子进程异步执行，所以看起来还是串行顺序。
2. 未等待进程执行完成：启动进程后立刻执行文件写入，此时子进程可能还在运行，根本拿不到有效结果。
3. 结果传递与写入逻辑错误：原关键字方法没有返回值，且Process对象不能直接作为writerows的参数，会触发报错。
4. 缺失依赖模块：代码中使用csv模块但未导入，运行时会抛出模块未找到的错误。

修正后的完整代码 #

from robot.libraries.BuiltIn import BuiltIn
from multiprocessing import Process, Queue
import csv

class importABR:
    def __init__(self):
        pass

    def abr1_keyword(self, result_queue):
        # 每个子进程单独初始化BuiltIn实例，避免多进程状态冲突
        builtin = BuiltIn()
        builtin.import_resource('${EXECDIR}/Resources/HealthCheck.robot')
        # 执行keyword1并获取返回结果，假设返回可迭代格式（如列表嵌套列表）
        result = builtin.run_keyword('keyword1')
        result_queue.put(result)

    def aaa_radius(self, result_queue):
        builtin = BuiltIn()
        builtin.import_resource('${EXECDIR}/Resources/HealthCheck.robot')
        result = builtin.run_keyword('keyword2')
        result_queue.put(result)

    def custom_keyword(self, file):
        # 使用Queue实现多进程间的结果传递（多进程内存不共享，无法直接返回值）
        result_queue = Queue()
        
        # 正确创建子进程：target传方法引用，args传所需参数
        abr_process = Process(target=self.abr1_keyword, args=(result_queue,))
        radius_process = Process(target=self.aaa_radius, args=(result_queue,))
        
        # 启动子进程
        abr_process.start()
        radius_process.start()
        
        # 等待两个子进程全部执行完毕
        abr_process.join()
        radius_process.join()
        
        # 从队列中取出执行结果
        abr_result = result_queue.get()
        radius_result = result_queue.get()
        
        # 写入结果到文件
        with open(str(file), 'w', newline='') as out_file:
            writer = csv.writer(out_file)
            # 根据结果格式选择写入方式：多行用writerows，单行用writerow
            if isinstance(abr_result, (list, tuple)) and all(isinstance(item, (list, tuple)) for item in abr_result):
                writer.writerows(abr_result)
            else:
                writer.writerow([abr_result])
            if isinstance(radius_result, (list, tuple)) and all(isinstance(item, (list, tuple)) for item in radius_result):
                writer.writerows(radius_result)
            else:
                writer.writerow([radius_result])

关键修正说明 #

• 进程创建方式：target直接传递方法引用（如self.abr1_keyword），通过args参数传递子进程需要的参数（这里用Queue传递结果），避免方法提前执行。
• 结果收集：利用multiprocessing.Queue实现主进程与子进程的结果传递，解决多进程内存隔离导致的返回值无法直接获取问题。
• 等待进程完成：调用join()方法确保子进程全部执行完毕后再写入文件，避免写入空数据或不完整数据。
• Robot库安全调用：每个子进程单独初始化BuiltIn实例，防止多进程间的状态干扰。
• 文件写入兼容：增加结果格式判断，兼容单行和多行数据的写入场景，避免格式错误引发异常。

Robot Framework调用示例 #

在Robot测试用例中调用修正后的关键字：

*** Test Cases ***
Run Parallel Keywords
    importABR.custom_keyword    ${EXECDIR}/parallel_output.csv

内容的提问来源于stack exchange，提问作者Shubham Joshi