Flask에서 작업 스케줄러 설정

This post was originally published at https://rudra.dev/posts/setting-up-a-task-scheduler-in-flask

작업 스케줄러를 고려할 때 가장 먼저 떠오르는 것은 cron 작업입니다. 오늘날 대부분의 서버는 Linux 시스템에서 호스팅되므로 정기적인 작업을 위해 cron 작업을 설정하는 것이 많은 사람들에게 좋은 옵션처럼 보일 수 있습니다. 그러나 프로덕션에서 crontab을 사용하는 것은 a**의 고통일 뿐입니다. 서버 위치에 따라 다른 시간대를 구성하는 것은 약간 까다로울 수 있습니다.

이 접근 방식의 가장 큰 문제는 애플리케이션이 여러 웹 서버로 확장될 때입니다. 이 경우 하나를 실행하는 대신 경합 상태로 이어질 수 있는 여러 크론 작업을 실행할 수 있습니다. 또한 작업에 문제가 발생하면 디버깅하기 어렵습니다.

Flask에는 세 번째 문제를 해결하는 여러 가지 방법이 있으며 Celery은 가장 인기 있는 방법 중 하나입니다. 셀러리는 위의 문제를 매우 우아하게 해결합니다. 시간대를 계산하고 스케줄러 타이밍을 정확하게 설정하는 데 도움이 되는 pytz의 동일한 시간대를 사용합니다.

Celery는 백엔드 메시지 브로커(redis 또는 RabbitMQ)를 사용하여 서로 다른 웹 서버에서 실행되는 여러 셀러리 작업자를 위한 중앙 데이터베이스 서버 역할을 하는 일정 상태를 저장합니다. 메시지 브로커는 작업이 일정에 따라 한 번만 실행되도록 합니다. , 따라서 경쟁 조건을 제거합니다.

실시간 이벤트 모니터링도 Celery에서 지원됩니다. 여기에는 모든 현재 이벤트를 표시하는 아름다운 내장형 터미널 인터페이스가 포함되어 있습니다. 멋진 독립형 프로젝트Flower는 Celery 작업자 및 작업을 관리하는 웹 기반 도구를 제공합니다. 또한 장기 실행 작업에 유용한 비동기 작업 실행을 지원합니다.

해킹하러 가자

flask-celery
│   
│  app.py
│  docker-compose.yml
│  Dockerfile
│  entrypoint.sh
│  requirements.txt
│   
└────────────────────────

FROM python:3.7

# Create a directory named flask
RUN mkdir flask

# Copy everything to flask folder
COPY . /flask/

# Make flask as working directory
WORKDIR /flask

# Install the Python libraries
RUN pip3 install --no-cache-dir -r requirements.txt

EXPOSE 5000

# Run the entrypoint script
CMD ["bash", "entrypoint.sh"]

이 애플리케이션에 필요한 패키지는 requirements.txt 파일에 언급되어 있습니다.

Flask==1.0.2
celery==4.3.0
redis==3.3.11

진입점 스크립트는 여기로 이동합니다.

#!/bin/sh

flask run --host=0.0.0.0 --port 5000

Celery는 메시지 브로커를 사용하여 웹 앱과 셀러리 작업자 간에 메시지를 전달합니다. 여기서는 메시지 브로커로 사용될 Redis 컨테이너를 설정합니다.

version: "3.7"

services:

  redis:
    container_name: redis_dev_container
    image: redis
    ports:
      - "6379:6379"

  flask_service:
    container_name: flask_dev_container
    restart: always
    image: flask
    build: 
      context: ./
      dockerfile: Dockerfile
    depends_on:
        - redis
    ports:
      - "5000:5000"
    volumes:
      - ./:/flask
    environment:
        - FLASK_DEBUG=1

이제 작은 실험을 시작할 준비가 모두 끝났습니다. 포트 6379에서 실행되는 redis 컨테이너와 localhost:5000에서 실행되는 플라스크 컨테이너가 있습니다. 작은 웹 애플리케이션이 작동하는지 테스트하기 위해 간단한 API를 추가해 보겠습니다.

from flask import Flask

app = Flask(__name__)

@app.route("/")
def index_view():
    return "Flask-celery task scheduler!"

if __name__ == "__main__":
    app.run()

짜잔!



이제 응용 프로그램이 시작된 이후 경과 시간을 표시하는 간단한 타이머 응용 프로그램을 만들 것입니다. Redis 서버 URL로 셀러리를 구성해야 하며 시간을 저장하기 위해 다른 Redis 데이터베이스를 사용할 것입니다.

from flask import Flask
from celery import Celery
import redis

app = Flask(__name__)

# Add Redis URL configurations
app.config["CELERY_BROKER_URL"] = "redis://redis:6379/0"
app.config["CELERY_RESULT_BACKEND"] = "redis://redis:6379/0"

# Connect Redis db
redis_db = redis.Redis(
    host="redis", port="6379", db=1, charset="utf-8", decode_responses=True
)

# Initialize timer in Redis
redis_db.mset({"minute": 0, "second": 0})

# Add periodic tasks
celery_beat_schedule = {
    "time_scheduler": {
        "task": "app.timer",
        # Run every second
        "schedule": 1.0,
    }
}

# Initialize Celery and update its config
celery = Celery(app.name)
celery.conf.update(
    result_backend=app.config["CELERY_RESULT_BACKEND"],
    broker_url=app.config["CELERY_BROKER_URL"],
    timezone="UTC",
    task_serializer="json",
    accept_content=["json"],
    result_serializer="json",
    beat_schedule=celery_beat_schedule,
)


@app.route("/")
def index_view():
    return "Flask-celery task scheduler!"


@app.route("/timer")
def timer_view():
    time_counter = redis_db.mget(["minute", "second"])
    return f"Minute: {time_counter[0]}, Second: {time_counter[1]}"


@celery.task
def timer():
    second_counter = int(redis_db.get("second")) + 1
    if second_counter >= 59:
        # Reset the counter
        redis_db.set("second", 0)
        # Increment the minute
        redis_db.set("minute", int(redis_db.get("minute")) + 1)
    else:
        # Increment the second
        redis_db.set("second", second_counter)


if __name__ == "__main__":
    app.run()

Celery 작업자와 비트 서버를 모두 백그라운드 프로세스로 실행하도록 entrypoint.js를 업데이트해 보겠습니다.

#!/bin/sh

# Run Celery worker
celery -A app.celery worker --loglevel=INFO --detach --pidfile=''

# Run Celery Beat
celery -A app.celery beat --loglevel=INFO --detach --pidfile=''

flask run --host=0.0.0.0 --port 5000

우리만의 타이머

The application is only for demonstration purpose. The counter won't be accurate as the task processing time is not taken into account while calculating time.

모니터링 이벤트

Celery는 작업, 작업자 및 이벤트에 대한 다양한 통계를 모니터링하기 위한 풍부한 지원을 제공합니다. 이벤트를 활성화하고 모니터링하려면 컨테이너에 로그인해야 합니다.

docker exec -it flask_dev_container bash

모든 이벤트 활성화 및 나열

celery -A app.celery control enable_events

celery -A app.celery control events

이렇게 하면 예약된 작업의 모든 세부 정보를 나열하는 멋진 대화형 터미널 UI가 표시됩니다.

결론

이 게시물에서는 Celery의 주요 목적이 작업 대기열을 처리하는 것이지만 crontab의 더 나은 대안으로 Celery를 사용했습니다. 웹 컨테이너에서 백그라운드 프로세스를 실행하는 것은 모범 사례로 간주되지 않으므로 Celery 작업자와 비트 서버는 서로 다른 컨테이너에서 실행할 수 있습니다.

어리석은 타이머 응용 프로그램을 만들지 않는 한.

위에서 언급한 코드는 여기에서 찾을 수 있습니다. repo

안녕!