python - 在执行 pool.map 时，是否可以在 python 中每 x 秒执行一次函数？

Question

我在大数据阵列上运行 pool.map，我想每分钟在控制台中打印报告。是否可以？据我了解，python是同步语言，它不能像nodejs那样做到这一点。

也许它可以通过线程来完成......或者如何？

finished = 0

def make_job():
   sleep(1)
   global finished
   finished += 1

# I want to call this function every minute
def display_status():
   print 'finished: ' + finished

def main():
    data = [...]
    pool = ThreadPool(45)
    results = pool.map(make_job, data)
    pool.close()
    pool.join()

Answer 1

您可以使用永久线程计时器，就像这个问题中的计时器:Python threading.timer - repeat function every 'n' seconds

from threading import Timer,Event 

class perpetualTimer(object):

   # give it a cycle time (t) and a callback (hFunction) 
   def __init__(self,t,hFunction):
      self.t=t
      self.stop = Event()
      self.hFunction = hFunction
      self.thread = Timer(self.t,self.handle_function)

   def handle_function(self):
      self.hFunction()
      self.thread = Timer(self.t,self.handle_function)
      if not self.stop.is_set():
          self.thread.start()

   def start(self):
      self.stop.clear()
      self.thread.start()

   def cancel(self):
      self.stop.set()
      self.thread.cancel()

基本上，这只是一个 Timer 对象的包装器，它会在每次调用所需函数时创建一个新的 Timer 对象。不要指望它能达到毫秒级的精度(甚至接近)，但对于您的目的而言，它应该是理想的。

使用这个你的例子会变成:

finished = 0

def make_job():
   sleep(1)
   global finished
   finished += 1

def display_status():
   print 'finished: ' + finished

def main():
    data = [...]
    pool = ThreadPool(45)

    # set up the monitor to make run the function every minute
    monitor = PerpetualTimer(60,display_status)
    monitor.start()
    results = pool.map(make_job, data)
    pool.close()
    pool.join()
    monitor.cancel()

编辑:

更清洁的解决方案可能是(感谢下面的评论):

from threading import Event,Thread 

class RepeatTimer(Thread):
    def __init__(self, t, callback, event):
        Thread.__init__(self)
        self.stop = event
        self.wait_time = t
        self.callback = callback
        self.daemon = True

    def run(self):
        while not self.stop.wait(self.wait_time):
            self.callback()

然后在你的代码中:

def main():
    data = [...]
    pool = ThreadPool(45)
    stop_flag = Event()
    RepeatTimer(60,display_status,stop_flag).start()
    results = pool.map(make_job, data)
    pool.close()
    pool.join()
    stop_flag.set()