如何在 Python 中使用多处理定期调用函数？答案

【问题标题】：How to periodically call a function using multiprocessing in Python?如何在 Python 中使用多处理定期调用函数？
【发布时间】：2019-08-01 17:40:32
【问题描述】：

我想调用一个函数，该函数使用多处理库定期检查总线上的值（需要非阻塞）。有一种使用线程库的方法，但它不使用多个进程。

我已经看到使用线程库的此类功能的示例代码，但我想使用多处理库来实现相同的功能。多处理状态的官方文档：

“注意多处理不包含 threading.active_count()、threading.enumerate()、threading.settrace()、threading.setprofile()、threading.Timer 的类似物”

但是，在我看到的线程库示例中，它们使用 threading.Timer。多处理有类似的功能吗？

import time, threading
def foo():
    print(time.ctime())
    threading.Timer(10, foo).start()

foo()

#output:
#Thu Dec 22 14:46:08 2011
#Thu Dec 22 14:46:18 2011
#Thu Dec 22 14:46:28 2011
#Thu Dec 22 14:46:38 2011

Executing periodic actions in Python

上面的代码是与线程库一起使用的示例。另外，我想知道这是否是不好的做法，因为线程永远不会终止（.join()）。

【问题讨论】：

@wwii 这不是重复的，因为我想使用多处理库函数来执行此操作。我不想使用线程库进行定期函数调用。

标签： python python-multiprocessing

【解决方案1】：

基本上，多处理模块中没有计时器类。应该是，但似乎当他们从 pyprocessing 迁移到 multiprocessing 时，他们没有包括那部分。这里解释了https://stackoverflow.com/a/25297863/6598433。

您可以像https://stackoverflow.com/a/25297758/6598433 中发布的 dano 一样手动为多处理库创建一个工作计时器

from multiprocessing import Process, Event

class Timer(Process):
    def __init__(self, interval, function, args=[], kwargs={}):
        super(Timer, self).__init__()
        self.interval = interval
        self.function = function
        self.args = args
        self.kwargs = kwargs
        self.finished = Event()

    def cancel(self):
        """Stop the timer if it hasn't finished yet"""
        self.finished.set()

    def run(self):
        self.finished.wait(self.interval)
        if not self.finished.is_set():
            self.function(*self.args, **self.kwargs)
        self.finished.set()

在这里查看完整的问题：Why no Timer class in Python's multiprocessing module?

【讨论】：

谢谢。我会试试这个实现。
进行定期调用的最佳方法是使用具有 while 循环的 run() 函数。在 while 循环结束时，您添加一个函数，该函数在 while 循环开始时检查设定的时间段和初始时间。然后函数等待当前时间和开始时间之间的差，如果周期大于那个时间。所以if( time.time() - start_time > desired_period) { time.sleep( desired_period-(time.time()-start_time) ) }

【解决方案2】：

要扩展 Nosvan 的答案以拥有一个真正的周期性计时器（它会漂移几毫秒），您只需像这样扩展 Timer 类：

from typing import Callable


class PeriodicTimer(Timer):
    def __init__(self, interval: int, function: Callable):
        super(PeriodicTimer, self).__init__(interval, function)

    def run(self):
        while not self.finished_event.is_set():
            # function callback could set the stop event
            self.function(*self.args, **self.kwargs)
            self.finished_event.wait(self.timeout)

我可以建议修改Timer 类，使其在构造函数中没有可变参数：

from multiprocessing import Process
from multiprocessing import Event
from typing import Callable


class Timer(Process):
    def __init__(self, timeout: int, function: Callable, args=None, kwargs=None):
        super(Timer, self).__init__()
        self.timeout = timeout
        self.function = function
        self.args = [] if args is None else args
        self.kwargs = {} if kwargs is None else kwargs
        self.finished_event = Event()

    def cancel(self):
        """Stop the timer if it hasn't finished yet"""
        self.finished_event.set()

    def run(self):
        self.finished_event.wait(self.timeout)
        if not self.finished_event.is_set():
            self.function(*self.args, **self.kwargs)
        self.finished_event.set()

以下是为什么要避免可变默认参数的详尽解释：

https://web.archive.org/web/20200221224620/http://effbot.org/zone/default-values.htm

【讨论】：