python - 从APScheduler收集 future

我有以下代码：

import pandas as pd
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime
from apscheduler.schedulers.blocking import BlockingScheduler


class FutureScheduler(object):

    def __init__():
        self.futures = []
        self.scheduler = BlockingScheduler()
        self.pool = ThreadPoolExecutor(5)
        self.full_frame = pd.DataFrame()

    def start(self):
        job = self.scheduler.add_job(self.add_future, 'cron', day_of_week='mon-fri', hour='8-15', minute='*')
        self.scheduler.start()
        self.flush_csvs()

    def add_future(self):
        self.futures.append(self.pool.submit(self.long_running_task))

    def flush_csvs(self):
        for future in as_completed(self.futures):
            results = future.result()
            self.full_frame = pd.concat((self.full_frame, results))
            self.full_frame.to_csv('results.csv')
            print "flushed... Queue size: %s" % len(self.futures)

    def long_running_task(self):
        #takes a while may or may not return before the next one is kicked off

所以我的问题是flush_csvs循环内的代码永远不会运行。在调用as_completed之前，我是否必须将所有期货添加到列表中？有没有办法让BlockingScheduler返回未来？我看到它返回Job，但在这种情况下，我希望它更像将来。

最佳答案

那是行不通的，因为调度程序会阻止主线程继续执行。这样可以防止flush_csvs被执行。

self.scheduler.start()
self.flush_csvs()

但是，这可能不是您想要的。 APScheduler在内部使用线程池，因此回调（self.long_running_task）已在单独的线程中执行。

您可以根据需要的工作线程数，通过APScheduler更改此线程池的配置，如果需要多个核（使用ProcessPoolExecutor而不是ThreadPoolExecutor）等，则还可以配置每个作业以执行所需的操作。例如，为每分钟运行一次以合并（仅运行一次）的作业配置策略，而不是在出现延迟的情况下背对背多次运行。

http://apscheduler.readthedocs.io/en/latest/userguide.html#choosing-the-right-scheduler-job-store-s-executor-s-and-trigger-s

关于python - 从APScheduler收集 future ，我们在Stack Overflow上找到一个类似的问题：https://stackoverflow.com/questions/44186743/