Heim > Backend-Entwicklung > Python-Tutorial > Wie installiere und verwende ich das Drittanbietermodul Apscheduler in Python?

Wie installiere und verwende ich das Drittanbietermodul Apscheduler in Python?

王林
Freigeben: 2023-05-08 10:04:17
nach vorne
1605 Leute haben es durchsucht

„apscheduler-Modul installieren“

      APScheduler verfügt über drei integrierte Planungssysteme, darunter:

Planung im Cron-Stil (optionale Start-/Endzeiten)

Intervallbasierte Ausführung (Jobs in gleichmäßigen Intervallen ausführen, optionale Start-/Endzeiten als na ja) )

Einmalige verzögerte Aufgabenausführung (führen Sie den Job einmal zu einem bestimmten Datum/zur angegebenen Uhrzeit aus)

  • Unterstützte Backend-Speicherjobs

    APScheduler kann Planungssysteme und Jobspeicher-Backends beliebig kombinieren und anpassen, sofern unterstützt Zu den Backend-Speicherjobs gehören:
  • Memory
  • SQLAlchemy

MongoDB

  • Redis

  • RethinkDB

  • ZooKe eper

  • APScheduler besteht aus vier Komponenten

  • Trigger enthalten die Planung Logik, und jeder Job hat seinen eigenen Trigger, um den nächsten Ausführungszeitpunkt zu bestimmen. Abgesehen von ihrer eigenen Anfangskonfiguration sind Trigger völlig zustandslos.

  • Jobspeicher (Jobspeicher) speichern geplante Jobs. Der Standardjobspeicher speichert den Job einfach im Speicher, und andere Jobspeicher speichern den Job in der Datenbank. Wenn ein Job in einem dauerhaften Jobspeicher gespeichert wird, werden die Daten des Jobs beim Laden serialisiert und deserialisiert. Der Jobspeicher kann den Planer nicht gemeinsam nutzen.

Executors übernehmen die Ausführung von Jobs. Sie tun dies normalerweise, indem sie bestimmte aufrufbare Objekte an einen Thread oder Prozesspool im Job senden. Wenn der Job abgeschlossen ist, benachrichtigt der Ausführende den Planer.

  • Scheduler (Scheduler) Die Konfiguration von Jobspeicher und -ausführern kann im Scheduler vorgenommen werden, z. B. das Hinzufügen, Ändern und Entfernen von Jobs. Je nach Anwendungsszenario können verschiedene Scheduler ausgewählt werden. Es stehen 7 Optionen zur Verfügung: BlockingScheduler, BackgroundScheduler, AsyncIOScheduler, GeventScheduler, TornadoScheduler, TwistedScheduler und QtScheduler.

  • Einführung in jede Komponente

    Trigger
  • Wenn Sie einen Job planen, müssen Sie einen Trigger für den Job auswählen, um zu beschreiben, wann der Job ausgelöst wird:

  • Datum : Einmalige Angabe des Datums;

Intervall: Wie oft innerhalb eines bestimmten Zeitraums ausgeführt werden soll;

cron: Kompatibel mit dem Linux-Crontab-Format, das leistungsstärkste.

  • Datum ist die einfachste Art der Planung und der Job wird nur einmal ausgeführt. Seine Parameter sind wie folgt:

    1.run_date
  • (datetime|str) – Das Ausführungsdatum oder die Ausführungszeit des Jobs
  • 2.timezone

    (datetime.tzinfo|str) – Geben Sie die Zeitzone an
  • job Speicher

Wenn Ihre Anwendung bei jedem Start Jobs neu erstellt, verwenden Sie daher den Standard-Jobspeicher (MemoryJobStore). Wenn Sie den Job jedoch auch dann beibehalten müssen, wenn der Scheduler neu startet oder die Anwendung abstürzt, sollten Sie Select verwenden den spezifischen Jobspeicher basierend auf Ihrer Anwendungsumgebung. Beispiel: Verwenden Sie Mongo oder SQLAlchemy JobStore (wird zur Unterstützung der meisten RDBMS verwendet). Wenn Ihre Anwendung

CPU-intensive Vorgänge

umfasst, können Sie ProcessPoolExecutor verwenden, um mehr CPU-Kerne zu verwenden. Sie können auch beide gleichzeitig verwenden, indem Sie ProcessPoolExecutor als zweiten Executor verwenden.


Wählen Sie den richtigen Planer.


BlockingScheduler: Wenn der Planer das einzige ist, was in Ihrer App ausgeführt wird app Wird während der Ausführung verwendet.

AsyncIOScheduler: Wird verwendet, wenn Ihr Programm Asyncio (ein asynchrones Framework) verwendet.

GeventScheduler: Wird verwendet, wenn Ihr Programm gevent (ein leistungsstarkes Python-Parallelitäts-Framework) verwendet.

TornadoScheduler: Wird verwendet, wenn Ihr Programm auf Tornado (einem Web-Framework) basiert.

TwistedScheduler: Verwenden Sie, wenn Ihr Programm Twisted (ein asynchrones Framework) verwendet.
  • QtScheduler: Kann verwendet werden, wenn Ihre Anwendung eine Qt-Anwendung ist.
  • apscheduler-Modul verwendet
  • , um Jobs hinzuzufügen.

    Es gibt zwei Möglichkeiten, einen neuen Job hinzuzufügen:
  • add_job, um einen Job hinzuzufügen;

    pip install apscheduler
    Nach dem Login kopieren

    decorator-Modus, um einen Job hinzuzufügen.
  • # -*- coding:utf-8 -*-
    from apscheduler.schedulers.blocking import BlockingScheduler
    import datetime
    
    def my_job1():
        print('my_job1 is running, Now is %s' % datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
    
    def my_job2():
        print('my_job2 is running, Now is %s' % datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
    
    sched = BlockingScheduler()
    # 每隔5秒运行一次my_job1
    sched.add_job(my_job1, 'interval', seconds=5, id='my_job1')
    # 每隔5秒运行一次my_job2
    sched.add_job(my_job2, 'cron', second='*/5', id='my_job2')
    sched.start()
    Nach dem Login kopieren
  • Entfernungsjob

  • Kein Entfernungsjob

    # -*- coding:utf-8 -*-
    from apscheduler.schedulers.blocking import BlockingScheduler
    import datetime
    
    sched = BlockingScheduler()
    
    # 每隔5秒运行一次my_job1
    @sched.scheduled_job('interval', seconds=5, id='my_job1')
    def my_job1():
        print('my_job1 is running, Now is %s' % datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
    
    # 每隔5秒运行一次my_job2
    @sched.scheduled_job('cron', second='*/5', id='my_job2')
    def my_job2():
        print('my_job2 is running, Now is %s' % datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
    
    sched.start()
    Nach dem Login kopieren

    Codeausführungsergebnis:
  • Wie installiere und verwende ich das Drittanbietermodul Apscheduler in Python?

    使用remove() 移除作业

    # -*- coding:utf-8 -*-
    from apscheduler.schedulers.blocking import BlockingScheduler
    import datetime
    
    def my_job(text=""):
        print(text, 'my_job1 is running, Now is %s' % datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
    
    sched = BlockingScheduler()
    job = sched.add_job(my_job, 'interval', seconds=2, args=['第一个作业'])
    job.remove()
    # #如果有多个任务序列的话可以给每个任务设置ID号,可以根据ID号选择清除对象,且remove放到start前才有效
    sched.add_job(my_job, 'interval', seconds=2, id='my_job_id', args=['第二个作业'])
    sched.start()
    Nach dem Login kopieren

    代码执行结果:

    Wie installiere und verwende ich das Drittanbietermodul Apscheduler in Python?

    使用remove_job()移除作业

    # -*- coding:utf-8 -*-
    from apscheduler.schedulers.blocking import BlockingScheduler
    import datetime
    
    def my_job(text=""):
        print(text, 'my_job1 is running, Now is %s' % datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
    
    sched = BlockingScheduler()
    job = sched.add_job(my_job, 'interval', seconds=2, args=['第一个作业'])
    # #如果有多个任务序列的话可以给每个任务设置ID号,可以根据ID号选择清除对象,且remove放到start前才有效
    sched.add_job(my_job, 'interval', seconds=2, id='my_job_id', args=['第二个作业'])
    sched.remove_job('my_job_id')
    sched.start()
    Nach dem Login kopieren

    代码执行结果:

    Wie installiere und verwende ich das Drittanbietermodul Apscheduler in Python?

    触发器类型

    APScheduler有3中内置的触发器类型:

    • 新建一个调度器(scheduler);

    • 添加一个调度任务(job store);

    • 运行调度任务。

    代码实现

    # -*- coding:utf-8 -*-
    import time
    import datetime
    from apscheduler.schedulers.blocking import BlockingScheduler
    
    def my_job(text="默认值"):
        print(text, time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time())))
    
    sched = BlockingScheduler()
    sched.add_job(my_job, 'interval', seconds=3, args=['3秒定时'])
    # 2018-3-17 00:00:00 执行一次,args传递一个text参数
    sched.add_job(my_job, 'date', run_date=datetime.date(2019, 10, 17), args=['根据年月日定时执行'])
    # 2018-3-17 13:46:00 执行一次,args传递一个text参数
    sched.add_job(my_job, 'date', run_date=datetime.datetime(2019, 10, 17, 14, 10, 0), args=['根据年月日时分秒定时执行'])
    # sched.start()
    """
    interval 间隔调度,参数如下:
        weeks (int) – 间隔几周
        days (int) – 间隔几天
        hours (int) – 间隔几小时
        minutes (int) – 间隔几分钟
        seconds (int) – 间隔多少秒
        start_date (datetime|str) – 开始日期
        end_date (datetime|str) – 结束日期
        timezone (datetime.tzinfo|str) – 时区
    """
    """
    cron参数如下:
        year (int|str) – 年,4位数字
        month (int|str) – 月 (范围1-12)
        day (int|str) – 日 (范围1-31)
        week (int|str) – 周 (范围1-53)
        day_of_week (int|str) – 周内第几天或者星期几 (范围0-6 或者 mon,tue,wed,thu,fri,sat,sun)
        hour (int|str) – 时 (范围0-23)
        minute (int|str) – 分 (范围0-59)
        second (int|str) – 秒 (范围0-59)
        start_date (datetime|str) – 最早开始日期(包含)
        end_date (datetime|str) – 最晚结束时间(包含)
        timezone (datetime.tzinfo|str) – 指定时区
    """
    # my_job将会在6,7,8,11,12月的第3个周五的1,2,3点运行
    sched.add_job(my_job, 'cron', month='6-8,11-12', day='3rd fri', hour='0-3')
    # 截止到2018-12-30 00:00:00,每周一到周五早上五点半运行job_function
    sched.add_job(my_job, 'cron', day_of_week='mon-fri', hour=5, minute=30, end_date='2018-12-31')
    
    # 表示2017年3月22日17时19分07秒执行该程序
    sched.add_job(my_job, 'cron', year=2017, month=3, day=22, hour=17, minute=19, second=7)
    
    # 表示任务在6,7,8,11,12月份的第三个星期五的00:00,01:00,02:00,03:00 执行该程序
    sched.add_job(my_job, 'cron', month='6-8,11-12', day='3rd fri', hour='0-3')
    
    # 表示从星期一到星期五5:30(AM)直到2014-05-30 00:00:00
    sched.add_job(my_job, 'cron', day_of_week='mon-fri', hour=5, minute=30, end_date='2014-05-30')
    
    # 表示每5秒执行该程序一次,相当于interval 间隔调度中seconds = 5
    sched.add_job(my_job, 'cron', second='*/5', args=['5秒定时'])
    sched.start()
    Nach dem Login kopieren
    cron表达式参数描述
    *anyFire on every value
    */aanyFire every a values, starting from the minimum
    a-banyFire on any value within the a-b range (a must be smaller than b)
    a-b/canyFire every c values within the a-b range
    xth ydayFire on the x -th occurrence of weekday y within the month
    last xdayFire on the last occurrence of weekday x within the month
    lastdayFire on the last day within the month
    x,y,zanyFire on any matching expression; can combine any number of any of the above expressions

    使用SQLAlchemy作业存储器存放作业

    # -*- coding:utf-8 -*-
    from apscheduler.schedulers.blocking import BlockingScheduler
    from datetime import datetime
    import logging
    sched = BlockingScheduler()
    
    def my_job():
        print('my_job is running, Now is %s' % datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
    
    # 使用sqlalchemy作业存储器
    # 根据自己电脑安装的库选择用什么连接 ,如pymysql   其中:scrapy表示数据库的名称,操作数据库之前应创建对应的数据库
    url = 'mysql+pymysql://root:123456@localhost:3306/scrapy?charset=utf8'
    sched.add_jobstore('sqlalchemy', url=url)
    # 添加作业
    sched.add_job(my_job, 'interval', id='myjob', seconds=5)
    
    log = logging.getLogger('apscheduler.executors.default')
    log.setLevel(logging.INFO)  # DEBUG
    # 设定日志格式
    fmt = logging.Formatter('%(levelname)s:%(name)s:%(message)s')
    h = logging.StreamHandler()
    h.setFormatter(fmt)
    log.addHandler(h)
    
    sched.start()
    Nach dem Login kopieren

    暂停和恢复作业

    # 暂停作业:
    apsched.job.Job.pause()
    apsched.schedulers.base.BaseScheduler.pause_job()
    # 恢复作业:
    apsched.job.Job.resume()
    apsched.schedulers.base.BaseScheduler.resume_job()
    Nach dem Login kopieren

    获得job列表

    • get_jobs(),它会返回所有的job实例;

    • 使用print_jobs()来输出所有格式化的作业列表;

    • get_job(job_id=“任务ID”)获取指定任务的作业列表。

    代码实现:

    # -*- coding:utf-8 -*-
    from apscheduler.schedulers.blocking import BlockingScheduler
    import datetime
    
    def my_job(text=""):
        print(text, 'my_job1 is running, Now is %s' % datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
    
    sched = BlockingScheduler()
    
    job = sched.add_job(my_job, 'interval', seconds=2, args=['第一个作业'])
    sched.add_job(my_job, 'interval', seconds=2, id='my_job_id', args=['第二个作业'])
    
    print(sched.get_jobs())
    
    print(sched.get_job(job_id="my_job_id"))
    
    sched.print_jobs()
    sched.start()
    Nach dem Login kopieren

    关闭调度器

    默认情况下调度器会等待所有正在运行的作业完成后,关闭所有的调度器和作业存储。如果你不想等待,可以将wait选项设置为False。

    sched.shutdown()
    sched.shutdown(wait=False)
    Nach dem Login kopieren

Das obige ist der detaillierte Inhalt vonWie installiere und verwende ich das Drittanbietermodul Apscheduler in Python?. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!

Verwandte Etiketten:
Quelle:yisu.com
Erklärung dieser Website
Der Inhalt dieses Artikels wird freiwillig von Internetnutzern beigesteuert und das Urheberrecht liegt beim ursprünglichen Autor. Diese Website übernimmt keine entsprechende rechtliche Verantwortung. Wenn Sie Inhalte finden, bei denen der Verdacht eines Plagiats oder einer Rechtsverletzung besteht, wenden Sie sich bitte an admin@php.cn
Beliebte Tutorials
Mehr>
Neueste Downloads
Mehr>
Web-Effekte
Quellcode der Website
Website-Materialien
Frontend-Vorlage