python - 进程间共享锁

Question

我尝试关注 this solution以及this solution但到目前为止还没有成功:

当我运行以下代码块时:

global manager
global lock
manager = Manager()
lock = manager.Lock()

class MyClass(object):

    def get_next_chunk(self, numberlist, chunks):
        for i in range(0, len(numberlist), chunks):
            yield numberlist[i:i + chunks]

    def multi_process(self, numberlist):
        procs = 5
        chunksize = 100
        with Pool(procs) as pool:
            pool.map(self.process_numberlist,
                  self.get_next_chunk(numberlist, chunksize))
        return self.running_total_list

    def process_numberlist(self, numberlist):
        temp_num_list = []
        temp_num_list = self.getnewNumbers()
        logger.debug("temp_num_list length: " + str(len(temp_num_list)))
        try:
            lock.acquire()
        except Exception as e:
            logger.error("Couldn't acquire lock")
            logger.error(e)
            traceback.format_exc()
            logger.error(sys.exc_info()[0])
        self.running_total_list = self.running_total_list + temp
        logger.debug("New running_total_list length: "
                    + str(len(self.running_total_list)))
        lock.release()
        break

我的日志输出如下:

[process_numberlist() ] temp_num_list length: 5
[process_numberlist() ] New running_total_list result set length: 5
[process_numberlist() ] temp_num_list length: 6
[process_numberlist() ] New running_total_list result set length: 6
[process_numberlist() ] temp_num_list length: 4
[process_numberlist() ] New running_total_list result set length: 9

当我认为我的预期输出应该是这样的:

[process_numberlist() ] temp_num_list length: 5
[process_numberlist() ] New running_total_list result set length: 5
[process_numberlist() ] temp_num_list length: 6
[process_numberlist() ] New running_total_list result set length: 11
[process_numberlist() ] temp_num_list length: 4
[process_numberlist() ] New running_total_list result set length: 15

编辑 - 尝试 2

根据 Aaron 的建议查看更新。现在收到“只能加入可迭代”错误

global manager
global lock

class MyClass(object):

    def get_next_chunk(self, numberlist, chunks):
        for i in range(0, len(numberlist), chunks):
            yield numberlist[i:i + chunks]

    def multi_process(self, numberlist):
        procs = 5
        chunksize = 100
        manager = Manager()
        lock = manager.Lock()
        with Pool(procs) as pool:
            func = partial(self.process_numberlist, lock)
            pool.map(function,
              self.get_next_chunk(numberlist, chunksize))
        return self.running_total_list

    def process_numberlist(self, numberlist, lock):
        temp_num_list = []
        temp_num_list = self.getnewNumbers()
        logger.debug("temp_num_list length: " + str(len(temp_num_list)))
        try:
             lock.acquire()
             self.running_total_list = self.running_total_list + temp_num_list
             logger.debug("New running_total_list length: "
                + str(len(self.running_total_list)))
             lock.release()
        except Exception as e:
            logger.error("Couldn't acquire lock")
            logger.error(e)
            traceback.format_exc()
            logger.error(sys.exc_info()[0])
        break

EDIT # 3 - getNewNumbers() 没有包含在这个玩具示例中，它只返回一个整数数组。希望有帮助

Answer 1

在我看来，您的主要目标是访问共享资源 (running_total_list)，这就是我特别关注它的原因。

在您的示例中，您使用了 Pool，而我使用了 Process。你可以看看 this article了解两者之间的核心差异，并决定哪个更适合您的用例。

我想到了这个关于如何在多个进程之间共享资源的快速示例。这应该可以让您了解如何从那里开始:

from multiprocessing import Process, Lock, Manager

def gen_numbers():
    import random
    return [i for i in range(random.randint(4,11))]

def process_numberlist(lock, shared_list, num):
    temp_num_list = gen_numbers()
    print("Proc %s: temp_num_list length: %s" %(num, len(temp_num_list)))

    try:
        lock.acquire()
        shared_list += temp_num_list
        print("Proc %s: New shared_list length: %s" %(num, len(shared_list)))
    finally:
        lock.release()

lock = Lock()
manager = Manager()
shared_list = manager.list()

proc = 5
proc_list = []

for num in range(proc):
    p = Process(target=process_numberlist, args=( lock, shared_list, num+1, ))
    p.start()

    proc_list.append( p )

for p in proc_list:
    p.join()

需要注意的一件重要事情是这里定义了一个shared_list。与线程不同，每个进程都有自己的内存空间(Pool 也不异常(exception))，这就是它们之间共享数据不起作用的原因。这意味着，您需要实现某种进程间通信 (IPC)，幸运的是 python 已经为您提供了一些工具。其中之一是 multiprocessing.Manager。它公开了一些数据结构(如 dict 或 list)，您可以使用这些数据结构在进程之间共享。

在这种情况下，Lock 也是如此。这很重要，因为您不想同时从多个进程访问共享内存。这只会让您的程序变得不可预测。

另一件需要注意的事情是，process_numberlist 的执行不一定按顺序，因为每个进程彼此独立运行，但它们都可以访问相同的资源。

希望对您有所帮助!