python调度框架APScheduler使用的实例详解

本篇文章主要介绍了详解python调度框架APScheduler使用，小编觉得挺不错的，现在分享给大家，也给大家做个参考。一起跟随小编过来看看吧

最近在研究python调度框架APScheduler使用的路上，那么今天也算个学习笔记吧！

# coding=utf-8
"""
Demonstrates how to use the background scheduler to schedule a job that executes on 3 second
intervals.
"""
from datetime import datetime
import time
import os
from apscheduler.schedulers.background import BackgroundScheduler

def tick():
  print('Tick! The time is: %s' % datetime.now())


if __name__ == '__main__':
  scheduler = BackgroundScheduler()
  scheduler.add_job(tick, 'interval', seconds=3)　　#间隔3秒钟执行一次
  scheduler.start()  #这里的调度任务是独立的一个线程
  print('Press Ctrl+{0} to exit'.format('Break' if os.name == 'nt' else 'C'))

  try:
    # This is here to simulate application activity (which keeps the main thread alive).
    while True:
      time.sleep(2)  #其他任务是独立的线程执行
      print('sleep!')
  except (KeyboardInterrupt, SystemExit):
    # Not strictly necessary if daemonic mode is enabled but should be done if possible
    scheduler.shutdown()
    print('Exit The Job!')

非阻塞调度，在指定的时间执行一次

# coding=utf-8
"""
Demonstrates how to use the background scheduler to schedule a job that executes on 3 second
intervals.
"""

from datetime import datetime
import time
import os

from apscheduler.schedulers.background import BackgroundScheduler


def tick():
  print('Tick! The time is: %s' % datetime.now())


if __name__ == '__main__':
  scheduler = BackgroundScheduler()
  #scheduler.add_job(tick, 'interval', seconds=3)
  scheduler.add_job(tick, 'date', run_date='2016-02-14 15:01:05')　　#在指定的时间，只执行一次
  scheduler.start()  #这里的调度任务是独立的一个线程
  print('Press Ctrl+{0} to exit'.format('Break' if os.name == 'nt' else 'C'))

  try:
    # This is here to simulate application activity (which keeps the main thread alive).
    while True:
      time.sleep(2)  #其他任务是独立的线程执行
      print('sleep!')
  except (KeyboardInterrupt, SystemExit):
    # Not strictly necessary if daemonic mode is enabled but should be done if possible
    scheduler.shutdown()
    print('Exit The Job!')

非阻塞的方式，采用cron的方式执行

# coding=utf-8
"""
Demonstrates how to use the background scheduler to schedule a job that executes on 3 second
intervals.
"""

from datetime import datetime
import time
import os

from apscheduler.schedulers.background import BackgroundScheduler


def tick():
  print('Tick! The time is: %s' % datetime.now())

if __name__ == '__main__':
  scheduler = BackgroundScheduler()
  #scheduler.add_job(tick, 'interval', seconds=3)
  #scheduler.add_job(tick, 'date', run_date='2016-02-14 15:01:05')
  scheduler.add_job(tick, 'cron', day_of_week='6', second='*/5')
  '''
    year (int|str) – 4-digit year
    month (int|str) – month (1-12)
    day (int|str) – day of the (1-31)
    week (int|str) – ISO week (1-53)
    day_of_week (int|str) – number or name of weekday (0-6 or mon,tue,wed,thu,fri,sat,sun)
    hour (int|str) – hour (0-23)
    minute (int|str) – minute (0-59)
    second (int|str) – second (0-59)
    
    start_date (datetime|str) – earliest possible date/time to trigger on (inclusive)
    end_date (datetime|str) – latest possible date/time to trigger on (inclusive)
    timezone (datetime.tzinfo|str) – time zone to use for the date/time calculations (defaults to scheduler timezone)
  
    *  any  Fire on every value
    */a  any  Fire every a values, starting from the minimum
    a-b  any  Fire on any value within the a-b range (a must be smaller than b)
    a-b/c  any  Fire every c values within the a-b range
    xth y  day  Fire on the x -th occurrence of weekday y within the month
    last x  day  Fire on the last occurrence of weekday x within the month
    last  day  Fire on the last day within the month
    x,y,z  any  Fire on any matching expression; can combine any number of any of the above expressions
  '''
  scheduler.start()  #这里的调度任务是独立的一个线程
  print('Press Ctrl+{0} to exit'.format('Break' if os.name == 'nt' else 'C'))

  try:
    # This is here to simulate application activity (which keeps the main thread alive).
    while True:
      time.sleep(2)  #其他任务是独立的线程执行
      print('sleep!')
  except (KeyboardInterrupt, SystemExit):
    # Not strictly necessary if daemonic mode is enabled but should be done if possible
    scheduler.shutdown()
    print('Exit The Job!')

阻塞的方式，间隔3秒执行一次

# coding=utf-8
"""
Demonstrates how to use the background scheduler to schedule a job that executes on 3 second
intervals.
"""

from datetime import datetime
import os

from apscheduler.schedulers.blocking import BlockingScheduler


def tick():
  print('Tick! The time is: %s' % datetime.now())


if __name__ == '__main__':
  scheduler = BlockingScheduler()
  scheduler.add_job(tick, 'interval', seconds=3)
  
  print('Press Ctrl+{0} to exit'.format('Break' if os.name == 'nt' else 'C'))

  try:
    scheduler.start()  #采用的是阻塞的方式，只有一个线程专职做调度的任务
  except (KeyboardInterrupt, SystemExit):
    # Not strictly necessary if daemonic mode is enabled but should be done if possible
    scheduler.shutdown()
    print('Exit The Job!')

采用阻塞的方法，只执行一次

# coding=utf-8
"""
Demonstrates how to use the background scheduler to schedule a job that executes on 3 second
intervals.
"""

from datetime import datetime
import os

from apscheduler.schedulers.blocking import BlockingScheduler


def tick():
  print('Tick! The time is: %s' % datetime.now())


if __name__ == '__main__':
  scheduler = BlockingScheduler()
  scheduler.add_job(tick, 'date', run_date='2016-02-14 15:23:05')
  
  print('Press Ctrl+{0} to exit'.format('Break' if os.name == 'nt' else 'C'))

  try:
    scheduler.start()  #采用的是阻塞的方式，只有一个线程专职做调度的任务
  except (KeyboardInterrupt, SystemExit):
    # Not strictly necessary if daemonic mode is enabled but should be done if possible
    scheduler.shutdown()
    print('Exit The Job!')

采用阻塞的方式，使用cron的调度方法

# coding=utf-8
"""
Demonstrates how to use the background scheduler to schedule a job that executes on 3 second
intervals.
"""

from datetime import datetime
import os

from apscheduler.schedulers.blocking import BlockingScheduler


def tick():
  print('Tick! The time is: %s' % datetime.now())


if __name__ == '__main__':
  scheduler = BlockingScheduler()
  scheduler.add_job(tick, 'cron', day_of_week='6', second='*/5')
  '''
    year (int|str) – 4-digit year
    month (int|str) – month (1-12)
    day (int|str) – day of the (1-31)
    week (int|str) – ISO week (1-53)
    day_of_week (int|str) – number or name of weekday (0-6 or mon,tue,wed,thu,fri,sat,sun)
    hour (int|str) – hour (0-23)
    minute (int|str) – minute (0-59)
    second (int|str) – second (0-59)
    
    start_date (datetime|str) – earliest possible date/time to trigger on (inclusive)
    end_date (datetime|str) – latest possible date/time to trigger on (inclusive)
    timezone (datetime.tzinfo|str) – time zone to use for the date/time calculations (defaults to scheduler timezone)
  
    *  any  Fire on every value
    */a  any  Fire every a values, starting from the minimum
    a-b  any  Fire on any value within the a-b range (a must be smaller than b)
    a-b/c  any  Fire every c values within the a-b range
    xth y  day  Fire on the x -th occurrence of weekday y within the month
    last x  day  Fire on the last occurrence of weekday x within the month
    last  day  Fire on the last day within the month
    x,y,z  any  Fire on any matching expression; can combine any number of any of the above expressions
  '''
  
  print('Press Ctrl+{0} to exit'.format('Break' if os.name == 'nt' else 'C'))

  try:
    scheduler.start()  #采用的是阻塞的方式，只有一个线程专职做调度的任务
  except (KeyboardInterrupt, SystemExit):
    # Not strictly necessary if daemonic mode is enabled but should be done if possible
    scheduler.shutdown()
    print('Exit The Job!')

Atas ialah kandungan terperinci python调度框架APScheduler使用的实例详解. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!