多处理任务未运行

Question

我有两个线程接收异步数据，并希望在主线程中处理这些数据。主线程调用 (ProcessA) 运行的第一个进程，但第二个从未执行。为了便于空间和解释，我简化了线程。 第一个线程是：

import logging
import time
logging.basicConfig(filename='ProcessA.log',level=logging.DEBUG)
from multiprocessing import Queue
q = Queue()
class ProcessA(q):
    global variableA 
    logging.info('ProcessA started') 
    while (True):
        variableA = 'A' + str(time.time()) 
        q.put(variableA)
        logging.info (variableA)
        time.sleep(5)

第二个线程类似：

import time
from multiprocessing import Queue
import logging
logging.basicConfig(filename='ProcessB.log',level=logging.DEBUG)
q = Queue()
class ProcessB(q):
    logging.info('ProcessB started') 
    global variableB 
    while (True):
        variableB = 'B' + str(time.time()) 
        q.put (variableB)
        logging.info (variableB)
        time.sleep(2)

主调用线程如下：

import time 
from multiprocessing import  Process, Queue
import ProcessA.py
import ProcessB.py

import logging
logging.basicConfig(filename='MThread.log',level=logging.DEBUG)

if __name__ == "__main__":
    queue = Queue()
    a =Process(target=ProcessA, args=(queue,))
    a.start()

    b = Process(target=ProcessB, args=(queue,))
    b.start()

    while (True):
        if not queue.empty():
            variableC = queue.get()
            logging.info ("variableC ="+ variableC) 
            time.sleep(1)

当我运行程序时，只有 ProcessA 运行（正如我在日志文件中看到的那样）。

无论我是在 Windows 7 终端中运行本机代码还是在 Spyder 中运行“运行>配置..>在外部终端中执行”设置，我都会得到相同的结果

我该怎么做才能让 ProcessB 和主线程的剩余部分运行？

更新 将类从外部文件移动到内部函数是可行的，但是使用外部函数是行不通的。

这就是这段代码的工作原理：

import time 
from multiprocessing import  Process, Queue
import logging
logging.basicConfig(filename='MThread2.log',level=logging.DEBUG)

def ProcessA(q):
    global variableA 
    logging.info('ProcessA started') 
    while (True):
        variableA = 'A' + str(time.time()) 
        q.put(variableA)
        logging.info (variableA)
        time.sleep(5)

def ProcessB(q):
    logging.info('ProcessB started') 
    global variableB 

    while (True):
        variableB = 'B' + str(time.time()) 
        q.put (variableB)
        logging.info (variableB)
        time.sleep(2)


if __name__ == "__main__":
    queueA = Queue()
    a =Process(target=ProcessA, args=(queueA,))
    a.start()
    queueB = Queue()
    b = Process(target=ProcessB, args=(queueB,))
    b.start()

    while (True):
        if not queueA.empty():
            variableC = queueA.get()
            logging.info ("variableC ="+ variableC) 
        if not queueB.empty():
            variableC = queueB.get()
            logging.info ("variableC ="+ variableC) 


        time.sleep(1)

但是，将代码移到外部函数中仍然不起作用。

Answer 1

这里很难说到底出了什么问题。我看到导入和分配你的类的一些小错误...

让你走上正确的道路：

ProcessA 和 B 类可以只是以 Queue 作为参数的函数，如果您想这样做的话。否则，它应该继承 Process（而不是 q）并在初始化时加入队列。这是一个很好的演练（队列和管道大约在一半）：https://www.geeksforgeeks.org/multiprocessing-python-set-2/

您也可以尝试使用管理器来共享此状态。见https://docs.python.org/3/library/multiprocessing.html#sharing-state-between-processes

Answer 2

问题在于我导入外部函数的方式。我不需要“.py”扩展名。我从这篇文章中发现了这一点：Call a function from another file in Python

以下代码有效：

import time 
from multiprocessing import  Process, Queue
from ProcessAF import processA
from ProcessBF import processB 


import logging
logging.basicConfig(filename='MThread.log',level=logging.DEBUG)

if __name__ == "__main__":

     queueA = Queue()
     a =Process(target=processA, args=(queueA,))
     a.start()

     queueB = Queue()
     b = Process(target=processB, args=(queueB,))
     b.start()

     while (True):
        if not queueA.empty():
            variableC = queueA.get()
            logging.info ("variableC ="+ variableC) 
            if not queueB.empty():
                variableC = queueB.get()
                logging.info ("variableC ="+ variableC) 


            time.sleep(1)

在同目录下一个名为ProcessAF.py的文件中的函数如下：

import logging
import time
logging.basicConfig(filename='ProcessA.log',level=logging.DEBUG)
from multiprocessing import Queue
q = Queue()
def processA(q):

    global variableA 
    logging.info('ProcessA started') 
    while (True):
        variableA = 'A' + str(time.time()) 
        q.put(variableA)
        logging.info (variableA)
       time.sleep(5)