[Python] scheduler 사용하기 #APScheduler

Python apscheduler

매달, 매달 특정일, 매일, 매일 특정 시간,분,초 등등

원하는 시간에 python script 를 실행하고 싶을 때 사용합니다.

대표적인것으로 schedule 와 apscheduler 가 있는데

apscheduler 사용법을 알아보겠습니다.

pip install APScheduler

자주 사용하는 BlockingScheduler 와 BackgroundScheduler 를 설명하겠습니다.

 

소스로 이해하기 - BlockingScheduler

import time
from apscheduler.schedulers.blocking import BlockingScheduler

sched = BlockingScheduler()

# 매일 12시 30분에 실행
@sched.scheduled_job('interval', seconds=5, id='test_1')
def job1():
    print(f'job1 : {time.strftime("%H:%M:%S")}')

# 매일 12시 30분에 실행
@sched.scheduled_job('cron', hour='12', minute='30', id='test_2')
def job2():
    print(f'job2 : {time.strftime("%H:%M:%S")}')

# 이런식으로 추가도 가능. 매분에 실행
sched.add_job(job2, 'cron', second='0', id="test_3")


print('sched before~')
sched.start()
print('sched after~') # 여긴 실행 안됨. Blocking 이기 때문에.

출력은

sched before~
job1 : 16:46:28
job1 : 16:46:33
# 이하 생략

"sched after~" 가 출력되지 않습니다.

sched.start() 에서 멈춰있는거죠. 그게 blocking 입니다.

 

BackgroundScheduler

sched 를 backgroundScheduler 로 바꿔보겠습니다.

위 소스에서 아래 부분만 바꿔주세요.

# from apscheduler.schedulers.blocking import BlockingScheduler
# sched = BlockingScheduler()
from apscheduler.schedulers.background import BackgroundScheduler
sched = BackgroundScheduler()

그리고 실행하면,

sched before~
sched after~

Process finished with exit code 0

프로그램이 바로 끝나버립니다.

sched.start() 다음에도 뭔가 작업을 해야한다면 Background 로 실행해야겠죠.

Background 를 테스트하려면 아래와 같은 방법으로 하면 됩니다.

import time
from apscheduler.schedulers.background import BackgroundScheduler
sched = BackgroundScheduler()

# 5초마다 실행
@sched.scheduled_job('interval', seconds=5, id='test_1')
def job1():
    print(f'job1 : {time.strftime("%H:%M:%S")}')

# 매일 12시 30분에 실행
@sched.scheduled_job('cron', hour='12', minute='30', id='test_2')
def job2():
    print(f'job2 : {time.strftime("%H:%M:%S")}')

# 이런식으로 추가도 가능. 매분에 실행
sched.add_job(job2, 'cron', second='0', id="test_3")


print('sched before~')
sched.start()
print('sched after~')

while True:
    time.sleep(1)

그럼 프로그램이 종료되지 않고 스캐줄 작업을 합니다.

 

파라메터 : id

작업 id 로 중복된 작업을 하지 않는데 사용합니다.

# 5초마다 실행
@sched.scheduled_job('interval', seconds=5, id='test_1')
def job1():
    print(f'job1 : {time.strftime("%H:%M:%S")}')
    i=0
    while i < 8:
        time.sleep(1)
        i+=1
    print('job1 end!')

5초에 한번씩 job1() 을 실행하는데, job1() 은 8초 후 종료하게 되어있습니다.
그럼 job1()을 실행하고 5초후에 job1()을 실행 하려고 할때 id 가 'test_1' 인것이 실행 중 이므로 추가실행을 안합니다.
출력 결과는 아래와 같습니다.

sched before~
sched after~
job1 : 17:04:26
Execution of job "job1 (trigger: interval[0:00:05], next run at: 2020-09-04 17:04:31 KST)" skipped: maximum number of running instances reached (1)
job1 end!
job1 : 17:04:36
Execution of job "job1 (trigger: interval[0:00:05], next run at: 2020-09-04 17:04:41 KST)" skipped: maximum number of running instances reached (1)
job1 end!
# 반복..

 

 

추가

아래와 같이 콤마로 여러 시간,분등을 설정할 수 있습니다.

# 매일 [9시15분, 9시45분, 15시15분, 15시45분]에 실행
@sched.scheduled_job('cron', hour='9,15', minute='15,45', id='test_2')
def job2():
    print(f'job2 : {time.strftime("%H:%M:%S")}')