Python Multiprocessing队列同步异常问题求助：限频HTTP客户端任务无法完成

我看了你的代码，问题主要出在参数传递错误和任务处理逻辑的设计缺陷上，导致队列里的任务根本没被正确执行，所以队列长度一直保持100。下面一步步给你分析和修复：

1. 致命的参数传递错误 #

你在manager里调用pool.apply_async(self._worker, queue)的时候犯了一个关键错误：apply_async的第二个参数args需要是一个元组，用来传递给目标函数的参数。而你直接传了queue，这会被Python当成可迭代对象拆包——队列里的每个元素（也就是你submit的(request, callback)元组）会被拆成两个参数传给_worker，但你的_worker只接受self和queue两个参数，这会导致worker进程抛出TypeError，任务直接失败，queue.task_done()也不会被执行，队列里的任务永远不会被标记为完成。

修正方法：把队列作为单个参数放进元组里传递：

pool.apply_async(self._worker, (queue,))

2. 调整任务处理逻辑（更合理的设计） #

不过其实你的逻辑可以更清晰：manager应该负责从队列取出任务，然后把具体的任务（request和callback）传给worker，而不是让worker自己去队列取。这样可以避免进程间队列访问的潜在竞争，也让计数逻辑更准确。

修改_manager_和_worker函数：

def _manager_(self, period, requests, pool, queue, rps, running):
    limited = False
    current_requests = 0
    last_clear = time()
    while running.value:
        try:
            # 非阻塞取任务，避免空队列时一直阻塞
            req, cb = queue.get_nowait()
        except queue.Empty:
            # 队列空的时候短暂休眠，减少CPU占用
            time.sleep(0.01)
            continue

        current_time = time()
        # 重置请求计数和限速标记
        if last_clear + period <= current_time:
            rps.value = current_requests / (current_time - last_clear)
            last_clear = current_time
            limited = False
            current_requests = 0

        if limited:
            # 达到限速，把任务放回队列，休眠后再尝试
            queue.put((req, cb))
            time.sleep(0.01)
            continue

        # 提交任务到进程池，完成后标记队列任务结束
        pool.apply_async(self._worker, args=(req, cb), callback=lambda _: queue.task_done())
        current_requests += 1

def _worker(self, req: Request, cb: Callable[[Response], None]):
    # 直接处理传入的任务，不需要操作队列
    res = requests.send(req.prepare())
    cb(res)

这里的变化：

• manager主动从队列取任务，避免worker直接操作队列
• 用get_nowait()非阻塞取任务，配合休眠减少空轮询的CPU消耗
• 达到限速时把任务放回队列，保证任务不会丢失
• 用apply_async的callback来调用queue.task_done()，确保任务完成后标记队列

3. 修正Process类的导入 #

你用了from multiprocessing.dummy import Process，这是线程类，而你的Pool是进程池（multiprocessing.Pool）。虽然线程和进程池可以交互，但为了避免混淆和潜在的同步问题，建议统一用进程类：

from multiprocessing import JoinableQueue, Pool, Value, Process  # 替换原来的dummy.Process

4. 修复join方法的资源释放顺序 #

你的join方法里的操作顺序有问题，应该先停止manager，再等待队列完成，最后关闭进程池：

def join(self):
    # 先停止manager循环
    self._running.value = False
    self._manager.join()

    # 等待队列所有任务完成
    self._queue.join()
    self._queue.close()

    # 优雅关闭进程池
    self._pool.close()
    self._pool.join()

另外，__del__方法里调用join不太可靠，因为Python的垃圾回收时机不确定，建议在测试代码里显式调用pool.join()。

5. 测试代码调整 #

修改测试代码，最后显式调用join：

if __name__ == '__main__':
    pool = HTTPWorkerPool(10, 1)
    for _ in range(100):
        pool.submit(Request(method='GET', url='https://httpbin.org/get'), cb)
    
    # 等待所有任务完成
    pool.join()
    
    # 或者保留你的休眠查看RPS，最后再join
    # for _ in range(10):
    #     sleep(1)
    #     print(pool.rps.value)
    # pool.join()

其他注意事项 #

• Value对象在进程间共享时，简单赋值是线程/进程安全的，但如果是复杂操作（比如current_requests +=1），建议用Lock来保证原子性，不过这里current_requests是在manager进程/线程里单独修改的，暂时没问题。
• 进程池的worker数量如果不指定，默认是CPU核心数，如果你需要更多的并发，可以手动设置processes参数。

这样修改后，队列里的任务会被正确处理，队列长度会逐渐减少到0，RPS也会符合你的限速设置。

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