Python多进程Queue.get()性能瓶颈问题及优化咨询

背景 #

我用Python实现了一个事件生成器，包含三个独立的Process模块：

• Input模块：在不同时间唤醒，将时间戳放入Input Queue；
• Event模块：从Input Queue获取时间戳，渲染事件后发送至Event Queue；
• Output模块：从Event Queue获取渲染后的事件，写入指定端点。

在高EPS（每秒事件数）场景下，Event模块存在性能问题，核心瓶颈是Queue.get()调用，该操作占据了大部分执行时间。当前Input Queue已预先填满，始终可执行get操作。

性能分析代码及数据 #

以下是带性能分析的简化代码：

spent_on_render = 0.0
spent_on_putting = 0.0
spent_on_getting = 0.0
spent_summary = 0.0

events_batch = []
last_flush = perf_counter()

while True:
    start = perf_counter()

    if (
        len(events_batch) >= FLUSH_AFTER_SIZE
        or (perf_counter() - last_flush) > FLUSH_AFTER_SECONDS
    ):
        putting_time = perf_counter()

        for event in events_batch:
            event_queue.put(event)

        spent_on_putting += perf_counter() - putting_time

        events_batch.clear()
        last_flush = perf_counter()

    try:
        get_time = perf_counter()
        timestamp = input_queue.get(
            block=False,
            timeout=0
        )
        spent_on_getting += perf_counter() - get_time
    except Empty:
        spent_summary += perf_counter() - start
        continue

    render_time = perf_counter()
    events_batch.append(event_plugin.produce(timestamp=timestamp))
    spent_on_render += perf_counter() - render_time
    

    spent_summary += perf_counter() - start

执行一段时间后的性能数据如下：

=================================
Spent on render: 12.241767558232823
Spent on putting: 0.326323675999447
Spent on getting: 22.14875863072848
Spent summary: 35.32571034637658

补充测试及疑问 #

测试1：批量读取Input Queue后再批量写入Event Queue #

timestamps = []
start = perf_counter()
qsize = input_queue.qsize()
for _ in range(qsize):
    timestamps.append(input_queue.get())

print('Size:', qsize)
print('Time spent after get:', perf_counter() - start)

for ts in timestamps:
    event_queue.put(ts)

print('Time spent after put:', perf_counter() - start)

输出：

Size: 1000000
Time spent after get: 1.935170126002049
Time spent after put: 2.2695002569998906

测试2：读取一个元素后立即写入Event Queue #

start = perf_counter()
qsize = input_queue.qsize()
for _ in range(qsize):
    ts = input_queue.get()
    event_queue.put(ts)

print('Size:', qsize)
print('Time spent after get and put:', perf_counter() - start)

输出：

Size: 1000000
Time spent after get and put: 16.109829995999462

存在两个疑问：

1. 为何上述两种测试方式的耗时差异如此巨大？
2. 在当前高EPS场景下，如何提升Event模块的性能？是否应该采用其他IPC方式？

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