Python 多进程队列 get() block

Question

我写了一个小代理实用程序(tcp 端口转发器)，现在正在尝试对其进行测试。所以我的测试:

def test_forwarding(self):
    route = self.config.routes[0]
    q = multiprocessing.JoinableQueue()

    proc_listen = multiprocessing.Process(
        target=lambda q: q.put(str(
            subprocess.Popen(["nc", "-l", str(route.sink[1])], stdout=subprocess.PIPE, shell=False).communicate()[0])),
        args=(q,))
    proc_listen.start()

    proc_write = multiprocessing.Process(
        target=lambda: subprocess.Popen("nc %s %s < ~/Workshop/port-forwarder/tests/test_data.txt" % route.source,
                                        stdout=subprocess.PIPE, shell=True))
    proc_write.start()

    proc_write.join(3)
    proc_write.terminate()

    proc_listen.join(3)
    proc_listen.terminate()
    q.join()
    self.assertEqual(open("test_data.txt", 'r').readline(), q.get())

但测试卡在断言中的 get() 调用上。完全搞砸了，尝试了所有的东西，比如加入/终止。任何帮助将不胜感激。

这是带有来源的 github 存储库:enter link description here

Answer 1

如果你被get阻止了，你可以

1. 使用 get_nowait()

2. 或使用 empty()检查侧面是否有一些元素。不为空则获取，否则跳过