Django+Celery环境下运行非周期性Telegram Bot任务的实现方案咨询

1. 给常驻Bot任务加单实例锁 #

避免任务被重复触发导致多实例冲突，可复用Django自带缓存实现：

from django.core.cache import cache
from celery.exceptions import MaxRetriesExceededError

@celery_app.task(name='apps.envc.tasks.TelegramBot', bind=True, max_retries=3)
def TelegramBot(self):
    lock_key = "telegram_bot_running_lock"
    # 锁有效期设为24小时，任务正常退出会自动释放
    if not cache.add(lock_key, "running", timeout=86400):
        return "已有运行中的Bot实例，任务终止"
    
    try:
        return start_bot()
    except Exception as e:
        # 异常时释放锁，间隔10秒自动重试
        cache.delete(lock_key)
        try:
            raise self.retry(exc=e, countdown=10)
        except MaxRetriesExceededError:
            return "Bot启动重试次数已达上限"
    finally:
        cache.delete(lock_key)

注意：如果你的缓存用的是本地内存而非Redis等共享缓存，需要换成跨进程共享的锁实现

2. 配置专用任务队列 #

避免Bot任务和现有周期任务抢占资源，给Bot任务分配独立队列：
修改任务装饰器指定队列：

@celery_app.task(name='apps.envc.tasks.TelegramBot', bind=True, max_retries=3, queue='telegram_bot_queue')

启动独立Celery Worker处理该队列，关闭任务超时限制，并发数设为1即可：

celery -A <你的Django项目名> worker -Q telegram_bot_queue --concurrency=1 --task-time-limit=None -n bot_worker@%h

原有处理周期任务的Worker启动命令保持不变即可，二者互不影响。

3. 触发任务运行 #

该任务不需要加入CELERY_BEAT_SCHEDULE配置（Beat仅用于调度周期任务），手动触发一次即可常驻运行：
进入Django Shell执行触发命令：

python manage.py shell

from apps.envc.tasks import TelegramBot
TelegramBot.delay()

如果需要部署时自动启动，可在部署脚本中新增上述触发逻辑，触发前可先判断缓存中是否已存在运行锁，避免重复触发。

4. 常用操作 #

• 停止Bot：直接停止专用的bot_worker进程即可，锁会自动释放
• 重启Bot：停止对应Worker后重新触发任务即可

内容的提问来源于stack exchange，提问作者dpv