[Python] APScheduler 스케쥴링 라이브러리

APScheduler

공식 문서

• Advanced Python Scheduler
• 한 번 또는 주기적으로 Python 코드가 실행되도록 일정을 관리하는 Python 라이브러리
• Scheduler 인스턴스 별도의 Thread로 실행됨
• Scheduler에 의해 실행되는 작업들은 Thread로 실행됨 (최대 10개의 Thread)
• 3가지 트리거 유형 존재
  • date: 특정 시점에 한번만 실행
  • interval: 일정 시간 간격으로 실행
  • cron: 하루 중 특정 시간에 주기적으로 작업
• 작업 추가/삭제/수정, 일시중지/재개/종료 가능
• 동시 실행 인스턴스 수 제한 가능 (max_instances 설정)
• 누락된 작업 실행 가능(misfire_grace_time 설정)

 

 

APScheduler 예제

from apscheduler.schedulers.background import BackgroundScheduler

def test_func(name):
    print(f"Hello {name}!")

## init scheduler
scheduler = BackgroundScheduler(timezone='Asia/Seoul')

## start scheduler
scheduler.start()

## add job
for i in range(100):
    task_id = f"task{i}"
    print(f"add_job: {task_id}")
    scheduler.add_job(func=test_func,
        trigger="interval",
        args=[f"user{i}"],
        seconds=10,
        id=task_id,
        misfire_grace_time=60)

## main
while True:
    pass

테스크 등록

테스크 실행

** misfire_grace_time 설정 안했을 시

• misfire_grace_time: 누락된 작업 실행 가능한 시간
• 기본값: undefined object
• 사용 가능한 Thread가 부족한 경우 다음과 같이 에러 발생
  Run time of job "test_func (trigger: interval[0:00:10], next run at: 2022-06-18 00:45:33 KST)" was missed by 0:00:01.233551

 

 

Thread 수 확인

다음과 같은 명령어를 통해 프로그램에서 사용하는 Thread의 개수를 확인할 수 있다.

ps -ef | grep py # python 프로세스의 PID 추출
ps hH p {PID} | wc -l # 해당 프로세스의 thread 개수 도출

ex) result: 12 (= 메인 쓰레드(1) + APScheduler 쓰레드(1) + 실행 테스크 최대 쓰레드(10))

 

 

회고

• 주기적으로 데이터를 수집하는 프로그램을 만들기위해 스케줄 관리에 용이한 APScheduler을 사용해보았다.
• 동일한 주기에 트리거 되는 작업들이 많다보니 사용가능한 Thread가 부족해 누락된 작업들이 있었다.
  misfire_grace_time을 1분 정도로 잡아주어 1분 내로 누락된 작업들을 실행할 수 있었다