AttributeError: Can't pickle local object 'computation.. function1 using multiprocessing queue

2022-01-21 00:00:00 python schedule queue multiprocessing scheduler

问题描述

我有以下使用调度程序和多处理模块的代码:

I have the following code using the scheduler and multiprocessing module:

def computation():
    def function1(q):
        while True:
            daydate = datetime.now()
            number = random.randrange(1, 215)
            print('Sent to function2: ({}, {})'.format(daydate, number))
            q.put((daydate, number))
            time.sleep(2)

    def function2(q):
        while True:
            date, number = q.get()
            print("Recevied values from function1: ({}, {})".format(date, number))
            time.sleep(2)

    if __name__ == "__main__":
        q = Queue()
        a = Process(target=function1, args=(q,))
        a.start()
        b = Process(target=function2, args=(q,))
        b.start()
        a.join()
        b.join()

schedule.every().monday.at("08:45").do(computation)
schedule.every().tuesday.at("08:45").do(computation)

while True:
    schedule.run_pending()
    time.sleep(1)

但是在执行代码时出现以下错误:

However while executing the code gives the following error:

AttributeError: Can't pickle local object 'computation..功能1

AttributeError: Can't pickle local object 'computation.. function1

还有:

OSError: [WinError 87] 参数不正确

OSError: [WinError 87] The parameter is incorrect

如何解决这个问题?我试图通过在模块的顶层定义一个函数来解决这个问题，如文档中所述(https://docs.python.org/2/library/pickle.html#what-c​​an-be-pickled-and-unpickled)但是它仍然给出同样的错误.

How does one solve this problem? I've tried to solve this by define a function at the top level of a module as stated in the documents (https://docs.python.org/2/library/pickle.html#what-can-be-pickled-and-unpickled) however it still gives the same error.

解决方案

嵌套函数不是在顶层定义的函数，所以这就是你得到错误的原因.您需要将 function1 和 function2 的定义移到外面的计算.

Nested functions are not functions defined at the top-level so that's why you get the error. You need to relocate the definition of function1 and function2 outside of computation.

按照您的编写方式，您的流程将立即开始，而不是在您安排它们运行的​​日期.这可能符合您的预期:

How you wrote it, your processes would start immediately instead of on the date you scheduled them to run. That probably does what you intended:

import os
import time
import random
from multiprocessing import Process, Queue
from threading import Thread
from datetime import datetime
import schedule


def function1(q):
    while True:
        daydate = datetime.now()
        number = random.randrange(1, 215)
        fmt = '(pid: {}) Sent to function2: ({}, {})'
        print(fmt.format(os.getpid(), daydate, number))
        q.put((daydate, number))
        time.sleep(2)


def function2(q):
    while True:
        date, number = q.get()
        fmt = "(pid: {}) Received values from function1: ({}, {})"
        print(fmt.format(os.getpid(), date, number))
        # time.sleep(2) no need to sleep here because q.get will block until
        # new items are available


def computation():
    q = Queue()
    a = Process(target=function1, args=(q,))
    a.start()
    b = Process(target=function2, args=(q,))
    b.start()
    a.join()
    b.join()


if __name__ == "__main__":

    # We are spawning new threads as a launching platform for
    # computation. Without it, the next job couldn't start before the last
    # one has finished. If your jobs always end before the next one should 
    # start, you don't need this construct and you can just pass 
    # ...do(computation)
    schedule.every().friday.at("01:02").do(
        Thread(target=computation).start
    )
    schedule.every().friday.at("01:03").do(
        Thread(target=computation).start
    )

    while True:
        schedule.run_pending()
        time.sleep(1)

就像现在一样，您的进程将在启动一次后永远运行.如果这不是您想要的，您必须考虑实施一些停止条件.

As it is now, your processes would run forever after started once. If that's not what you want, you have to think about implementing some stop condition.

相关文章