python Apscheduler 기반 정시 작업 구현

안내

작업 장면에서 이런 장면을 만났는데 정기적으로 캐시 인터페이스를 실행하여 동기화 장치 설정에 사용해야 한다.먼저 떠오르는 것은 리눅스의crontab입니다. 정기적이거나 일정 시간 간격으로 임무를 수행할 수 있습니다.그러나 이 시간 임무를 하나의 모듈로 파이썬 프로젝트에 집적하거나 지속화 작업을 원한다면, 분명히crontab는 적용되지 않습니다.Python의 APScheduler 모듈은 이런 문제를 잘 해결할 수 있기 때문에 이 글을 전문적으로 씁니다. 간단한 입문부터 APScheduler의 가장 기초적인 사용 장면과 지구화 임무에 대한 문제를 기록하고 마지막으로 다른 프레임워크와 결합하여 심층적으로 정시 임무 모듈을 맞춤형으로 설정합니다.

간단한 소개

먼저 Apscheduler 모듈에 포함된 네 가지 구성 요소에 대해 간단히 설명합니다.

Trigger 트리거

작업

Excutor 집행기

Scheduler 스케줄러

Apscheduler가 포함하는 몇 가지 개념에 대해 대략 이해했으니, 지금부터 간단한 예시를 살펴보자.

# -*- coding: utf-8 -*-

from apscheduler.schedulers.blocking import BlockingScheduler
import time


def hello():
  print(time.strftime("%c"))


if __name__ == "__main__":
  scheduler = BlockingScheduler()
  scheduler.add_job(hello, 'interval', seconds=5)
  scheduler.start()

예제 출력:

Thu Dec 3 16:01:20 2020
Thu Dec 3 16:01:25 2020
Thu Dec 3 16:01:30 2020
Thu Dec 3 16:01:35 2020
Thu Dec 3 16:01:40 2020
..........

이 간단한 예는 위에서 언급한 몇 가지 구성 요소로 운행 논리를 분석합니다.

우선 Scheduler 스케줄러입니다. 이 예시에서 사용하는 Blocking Scheduler 스케줄러는 공식 문서에서 Blocking Scheduler가 당신의 이 시간 작업 프로그램이 유일하게 실행되는 프로그램에 적합하다는 것을 설명합니다.다시 말하면 Blocking Scheduler 스케줄러는 차단 스케줄러로 프로그램이 이런 스케줄러를 실행할 때 프로세스가 막혀서 다른 작업을 수행할 수 없다

그 다음은 Job 작업과 트리거입니다. 이 두 가지를 함께 말하자면 작업을 정의할 때 트리거를 선택해야 하기 때문입니다. 여기서 선택한 것은 인터벌 트리거입니다. 이런 트리거는 고정된 시간 간격으로 작업을 실행합니다.다시 말하면 스케줄러에 Hello 작업을 추가하고 5초 간격으로 작업을 수행합니다

마지막은 실행기입니다. 기본값은ThreadPoolExcutor 실행기입니다. 작업에서 호출할 수 있는 대상을 스레드 탱크에 맡기고 작업을 수행하면 실행기가 스케줄러에게 알립니다.

내장형 Trigger 트리거 유형:

date: 특정 시간에 한 번만 작업을 실행합니다

interval: 고정된 시간 간격으로 작업을 한 번 실행합니다

cron: 하루 동안 특정 시간에 정기적으로 작업을 실행합니다.

일반적인 Scheduler 스케줄러:

Blocking Scheduler: 스케줄러는 프로세스에서 유일하게 실행되는 것입니다

BackgroundScheduler: 스케줄러는 프로그램 내부의 백그라운드에서 실행할 때 사용됩니다

AsyncIOScheduler: 응용 프로그램은 asyncio 모듈을 사용합니다

GeventScheduler: 응용 프로그램은 gevent 모듈을 사용합니다

TornadoScheduler: Tornado 응용 프로그램을 구축할 때 사용합니다

TwistedScheduler: Tornado 응용 프로그램을 구축할 때 사용합니다

QtScheduler: QT 응용 프로그램을 구축할 때 사용

일반적인 JobStore:

MemoryJobStore

MongoDBJobStore

SQLAlchemyJobStore

RedisJobStore

진급 사용

위의 간단한 예시를 통해 대략적인 작업 프로세스와 각 구성 요소가 전체 프로세스에서의 작용을 이해한다. 아래의 예는 Flask 웹 프레임워크와 Apscheduler 타이머를 결합하여 정해진 시간에 작업을 수행하는 것이다.

# -*- coding: utf-8 -*-

from flask import Flask, Blueprint, request
from apscheduler.executors.pool import ThreadPoolExecutor 
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.redis import RedisJobStore
import time

app = Flask(__name__)
executors = {"default": ThreadPoolExecutor(5)}
default_redis_jobstore = RedisJobStore(db=2, 
    jobs_key="apschedulers.default_jobs",
    run_times_key="apschedulers.default_run_times",
    host = '127.0.0.1',
    port = 6379
    )

scheduler = BackgroundScheduler(executors=executors)
scheduler.add_jobstore(default_redis_jobstore)
scheduler.start()

def say_hello():
  print(time.strftime("%c"))


@app.route("/get_job", methods=['GET'])
def get_job():
  if scheduler.get_job("say_hello_test"):
    return "YES"
  else:
    return "NO"

@app.route("/start_job", methods=["GET"])
def start_job():
  if not scheduler.get_job("say_hello_test"):
    scheduler.add_job(say_hello, "interval", seconds=5, id="say_hello_test")
    return "Start Scuessfully!"
  else:
    return "Started Failed"
  
@app.route("/remove_job", methods=["GET"])
def remove_job():
  if scheduler.get_job("say_hello_test"):
    scheduler.remove_job("say_hello_test")
    return "Delete Successfully!"
  else:
    return "Delete Failed"


if __name__ == "__main__":
  app.run(host="127.0.0.1", port=8787, debug=True)

먼저 Jobstore를 분석합니다. 여기서 사용하는 것은 Redis Jobstore입니다. 작업을 Redis 데이터베이스에 서열화합니다.여기에 참고로 작업 메모리를 설정해야 하는 이유는 스케줄러 프로그램이 붕괴되었을 때 작업을 보존할 수 있기 때문이다. 물론 어떤 작업 메모리를 선택하는지는 구체적인 작업 장면에 따라 현재 주류인 mysql,mongodb,redis,SQLite가 기본적으로 지원한다

그리고 다시 Scheduler를 보세요. 여기서 사용할 때 Background Scheduler는 스케줄러가 flask 프로그램의 정상적인 수신 요청을 막을 수 없기 때문에 Backgroun Scheduler에서 작업을 시작할 때 백그라운드에서 실행되도록 선택하고 메인 라인을 막지 않습니다

마지막으로 작업의 논리를 보자. 여기 get_job 작업 상태 얻기, 작업 존재 여부 보기,start_job는 작업이 시작되었는지 아닌지를 판단하고 시작 작업을 결정합니다.remove_job는 작업을 중지합니다.여기서 작업 정의는 인터벌 트리거를 통해 5초에 한 번씩say_hello 퀘스트;

총결산

마지막으로 요약하자면, 우선 스케줄러가 붕괴되어 다시 회복될 때, 작업 메모리에서 작업이 계속 실행될 수 있도록 작업 메모리를 설정해야 한다.그리고 작업 유형, 예를 들어 CPU 집약형 작업, 프로세스 풀 집행기를 설정해야 합니다. 기본적으로는 스레드 풀 집행기입니다.마지막으로 설정 스케줄러를 만들고 스케줄링을 시작합니다. 시작하기 전에 작업을 추가할 수도 있고, 시작한 후에 추가, 삭제, 작업을 가져올 수도 있습니다.(여기서 알아야 할 점은 응용 프로그램이 작업 메모리, 작업 또는 실행기를 직접 조작하지 않고 스케줄러가 적당한 인터페이스를 제공하여 이 인터페이스를 처리한다는 것이다.)
ApScheduler는 좋은 시간 작업 라이브러리로 동적으로 삭제를 추가할 수 있고 서로 다른 트리거 형식도 지원한다. 이것도 장점이다. 반대로 정적 작업이라면 linux의crontab 도구로 시간 작업을 할 수 있다.이 방면의 기록은 계속 갱신될 것이니 무슨 문제가 있으면 제기할 수 있으니 모두 함께 토론합시다.
이상은python Apscheduler의 사용 방법에 대한 상세한 내용입니다. 더 많은python Apscheduler에 대한 자료는 저희 다른 관련 글을 주목해 주십시오!