APScheduler cách thêm công việc ngoài bộ lập lịch?

Question

Tôi có một yêu cầu đơn giản. Tôi đang chạy apscheduler như là một quá trình riêng biệt. Tôi có một kịch bản lệnh jobproducer khác từ nơi tôi muốn thêm công việc vào bộ lập lịch biểu và chạy nó.

Đây là mã lên lịch của tôi,

# appsched.py 
from apscheduler.schedulers.blocking import BlockingScheduler 
scheduler = BlockingScheduler() 
scheduler.start() 

Đây là kịch bản sản xuất công việc của tôi,

# jobproducer.py 
from appsched import scheduler 

def say_hello_job(): 
    print "Hello" 

scheduler.add_job(say_hello_job, 'interval', minutes=1) 

Không cần phải nói rằng điều này không làm việc. Có cách nào để thực hiện công việc này bằng cách sử dụng một cửa hàng việc làm không? Cách chia sẻ lịch biểu với nhiều nhà sản xuất công việc khác nhau?

Answer 1

Tôi gặp sự cố tương tự, khi quá trình lên lịch của tôi là quy trình uWSGI MULE và có một ứng dụng riêng biệt mà tôi muốn thêm công việc mới.

Nhìn vào add_job() chức năng 's BaseScheduler:

with self._jobstores_lock: 
if not self.running: 
    self._pending_jobs.append((job, jobstore, replace_existing)) 
    self._logger.info('Adding job tentatively -- it will be properly scheduled when the scheduler starts') 
else: 
    self._real_add_job(job, jobstore, replace_existing, True) 

bạn sẽ nhìn thấy vấn đề: scheduler thêm công ăn việc làm chỉ khi nó đã được bắt đầu.

Giải pháp là may mắn khá đơn giản, chúng ta nên xác định riêng của chúng tôi "add-job-chỉ" Scheduler:

class JobAddScheduler(BlockingScheduler): 
    def add_job(self, func, trigger=None, args=None, kwargs=None, id=None, name=None, misfire_grace_time=undefined, 
       coalesce=undefined, max_instances=undefined, next_run_time=undefined, jobstore='default', 
       executor='default', replace_existing=False, **trigger_args): 

    job_kwargs = { 
     'trigger': self._create_trigger(trigger, trigger_args), 
     'executor': executor, 
     'func': func, 
     'args': tuple(args) if args is not None else(), 
     'kwargs': dict(kwargs) if kwargs is not None else {}, 
     'id': id, 
     'name': name, 
     'misfire_grace_time': misfire_grace_time, 
     'coalesce': coalesce, 
     'max_instances': max_instances, 
     'next_run_time': next_run_time 
    } 
    job_kwargs = dict((key, value) for key, value in six.iteritems(job_kwargs) if value is not undefined) 
    job = Job(self, **job_kwargs) 

    # Add jobs to job store 
    with self._jobstores_lock: 
     self._real_add_job(job, jobstore, replace_existing, True) 

    return job 

    def start(self): 
    pass 

    def shutdown(self, wait=True): 
    pass 

    def _main_loop(self): 
    pass 

    def wakeup(self): 
    pass 

Sau đó, chúng ta có thể thêm các công việc cron ngay lập tức:

jobscheduler = JobAddScheduler() 
jobscheduler.add_job(...) 

Don đừng quên cấu hình lịch trình! Trong trường hợp của tôi, tôi sử dụng SQLAlchemy-MySQL backend cho việc lưu trữ:

jobstores=dict(default=SQLAlchemyJobStore(url='mysql+pymsql://USER:PASSWORD@SERVER/DATABASE')) 
jobscheduler.configure(jobstores=jobstores) 

Tôi không chắc chắn về jobstores khác, nhưng sau khi tôi thêm một công việc mới, tôi đã phải gọi wakeup() chức năng của quá trình lên lịch riêng để "hoạt động" công việc. Tôi đã đạt được điều này bằng cách sử dụng hệ thống tín hiệu của uWSGI.