AiOpsWechat/celery_app.py

# from celery import Celery

# # 创建 Celery 应用
# celery_app = Celery(
#     'ai_ops_wechat_app',
#     broker='redis://:telpo%231234@192.168.2.121:8090/3', 
#     backend='redis://:telpo%231234@192.168.2.121:8090/3',
# )

# # 配置 Celery
# celery_app.conf.update(
#     task_serializer='json',
#     accept_content=['json'],
#     result_serializer='json',
#     timezone='Asia/Shanghai',
#     enable_utc=True,
# )

# #celery_app.autodiscover_tasks(['app.tasks'])

# from celery import Celery

# def make_celery(app):
#     celery = Celery(
#         app.import_name,
#         backend=app.config['CELERY_RESULT_BACKEND'],
#         broker=app.config['CELERY_BROKER_URL']
#     )
#     celery.conf.update(app.config)
#     # 自动发现任务
#     celery.autodiscover_tasks(['app.tasks'])
#     return celery

# # 初始化 Flask
# app = Flask(__name__)
# app.config.update(
#     CELERY_BROKER_URL='redis://:telpo%231234@192.168.2.121:8090/3',
#     CELERY_RESULT_BACKEND='redis://:telpo%231234@192.168.2.121:8090/3'
# )

# celery = make_celery(app)

from celery import Celery
import celery.schedules
from redbeat import RedBeatSchedulerEntry
from datetime import timedelta


from config import load_config,conf


load_config()

redis_host=conf().get("redis_host")
redis_port=conf().get("redis_port")
redis_password=conf().get("redis_password") 
redis_db=conf().get("redis_db") 

# 配置 Celery
celery_app = Celery(
    "worker",
    broker=f"redis://:telpo%231234@{redis_host}:{redis_port}/{redis_db}",
    backend=f"redis://:telpo%231234@{redis_host}:{redis_port}/{redis_db}",
    include=['tasks']
)

# 配置 redbeat 作为 Celery Beat 调度器
celery_app.conf.update(
    timezone="Asia/Shanghai",  # 设定时区
    beat_scheduler="redbeat.RedBeatScheduler",  # 使用 RedBeat 作为调度器
    redbeat_redis_url=f"redis://:telpo%231234@{redis_host}:{redis_port}/{redis_db}"  # redbeat 存储任务调度信息的 Redis
    ,
    redbeat_lock_timeout=60,  # 避免多个 Beat 实例冲突
    beat_max_loop_interval=5  # 让 Celery Beat 每 5 秒检查一次任务
)

task_name = "tasks.scheduled_task"
# 任务执行间隔（每 10 秒执行一次）
schedule = celery.schedules.schedule(timedelta(seconds=3))

# RedBeat 任务唯一 ID
redbeat_entry = RedBeatSchedulerEntry(
    name="redbeat:scheduled_task",  # 任务 ID
    task=task_name,  # 任务名称
    schedule=schedule,  # 任务调度时间
    args=[],
    app=celery_app
        
)

# 保存任务到 Redis
redbeat_entry.save()

# 自动发现任务
#celery.autodiscover_tasks(['tasks'])