Python：在multiprocess.Process中生成的两个线程之间共享变量

Question

Python 3.1.2

我遇到了multiprocessing.Process生成的两个线程之间的变量共享问题。这是一个简单的bool变量，它应该确定线程是否应该运行还是应该停止执行。下面是三种情况下显示的简化代码（但使用与原始代码相同的机制）：

1. 线程的主要类别。线程类型和self.is_running bool类型[工作正常]。
2. 多进程的主要类。进程类型和self.is_running bool类型[不工作。子线程具有self.is_running的本地副本，而不是共享它。
3. multiprocess的主类。过程类型和self.is_running属于multiprocessing.Value（“b”，True）[工作正常]。

我想要的是理解为什么它以这种方式工作而不是另一种方式。 （即为什么第2点不起作用，因为我假设它应该）。

测试是从python的解释器完成的：

from testclass import *

d = TestClass()
d.start()
d.stop()

以下是第1点的示例：

import threading
import time
import queue
import multiprocessing

class TestClass(threading.Thread):
def __init__(self):
    threading.Thread.__init__(self)
    self.q = queue.Queue(10)
    self.is_running = True
    self.sema = threading.Semaphore()

def isRunning(self):
    self.sema.acquire()
    print ("Am I running?", self.is_running)
    z = self.is_running
    self.sema.release()
    return z

def stop(self):
    self.sema.acquire()
    self.is_running = False
    print("STOPPING")
    self.sema.release()

def reader(self):
    while self.isRunning():
        print("R] Reading!")
        try:
            data = self.q.get(timeout=1)
        except:
            print("R] NO DATA!")
        else:
            print("R] Read: ", data)
def writer(self):
    while self.isRunning():
        print("W] Writing!")
        self.q.put(time.time())
        time.sleep(2)

def run(self):
    tr = threading.Thread(target=self.reader)
    tw = threading.Thread(target=self.writer)
    tr.start()
    tw.start()
    tr.join()
    tw.join()

---

第2点的例子：

import threading
import time
import queue
import multiprocessing


class Test(multiprocessing.Process):
def __init__(self):
    multiprocessing.Process.__init__(self)
    self.q = queue.Queue(10)
    self.is_running = True
    self.sema = threading.Semaphore()

def isRunning(self):
    self.sema.acquire()
    print ("Am I running?", self.is_running)
    z = self.is_running
    self.sema.release()
    return z

def stop(self):
    self.sema.acquire()
    self.is_running = False
    print("STOPPING")
    self.sema.release()

def reader(self):
    while self.isRunning():
        print("R] Reading!")
        try:
            data = self.q.get(timeout=1)
        except:
            print("R] NO DATA!")
        else:
            print("R] Read: ", data)
def writer(self):
    while self.isRunning():
        print("W] Writing!")
        self.q.put(time.time())
        time.sleep(2)

def run(self):
    tr = threading.Thread(target=self.reader)
    tw = threading.Thread(target=self.writer)
    tr.start()
    tw.start()
    tr.join()
    tw.join()

---

第3点的例子：

import threading
import time
import queue
import multiprocessing

class TestClass(multiprocessing.Process):
def __init__(self):
    multiprocessing.Process.__init__(self)
    self.q = queue.Queue(10)
    self.is_running = multiprocessing.Value("b", True)
    self.sema = threading.Semaphore()

def isRunning(self):
    self.sema.acquire()
    print ("Am I running?", self.is_running)
    z = self.is_running.value
    self.sema.release()
    return z

def stop(self):
    self.sema.acquire()
    self.is_running.value = False
    print("STOPPING")
    self.sema.release()

def reader(self):
    while self.isRunning():
        print("R] Reading!")
        try:
            data = self.q.get(timeout=1)
        except:
            print("R] NO DATA!")
        else:
            print("R] Read: ", data)
def writer(self):
    while self.isRunning():
        print("W] Writing!")
        self.q.put(time.time())
        time.sleep(2)

def run(self):
    tr = threading.Thread(target=self.reader)
    tw = threading.Thread(target=self.writer)
    tr.start()
    tw.start()
    tr.join()
    tw.join()

Answer 1

线程都是同一进程的一部分，因此它们共享内存。另一个结果是线程不能由不同的cpu同时完全执行，因为一个进程只能被一个cpu接收。

进程具有单独的内存空间。一个cpu可以运行一个进程，而另一个cpu运行另一个进程。让流程合作需要特殊的结构。

Answer 2

在第2点，父进程和子进程都有自己的is_running副本。当您在父进程中调用stop()时，它仅在父进程中修改is_running而不在子进程中修改。 multiprocessing.Value工作的原因是它的内存在两个进程之间共享。

如果需要进程感知队列，请使用multiprocessing.Queue。