### 避免 Flask 双活部署中 Celery 任务重复执行的解决方案 在 Flask 项目双活部署的情况下，为了避免 Celery 任务被重复执行，可以采取以下几种方法来确保任务的唯一性和一致性。这些方法包括使用分布式锁、任务去重机制以及合理的消息队列配置。 #### 1. 使用 Redis 分布式锁 通过 Redis 实现分布式锁，可以在多个实例之间协调任务的执行。只有成功获取锁的任务实例才会真正执行任务逻辑，其他实例则会跳过任务处理。这种方式可以有效避免任务重复执行的问题。 ```python import redis from celery import Celery celery = Celery('tasks', broker='redis://localhost:6379/0') redis_client = redis.StrictRedis(host='localhost', port=6379, db=0) @celery.task def async_send_message(openid, content): lock_key = f"lock:async_send_message:{openid}" with redis_client.lock(lock_key, timeout=10): # 设置锁超时时间 # 实际发送逻辑 pass ``` 这种方法的核心是通过 Redis 的 `SETNX` 命令实现分布式锁[^1]。 #### 2. Celery 内置的 `once` 插件 Celery 提供了 `celery-once` 插件，能够帮助开发者轻松实现任务去重。该插件支持基于 Redis 或 Memcached 的存储后端，并允许设置任务的唯一性规则。 安装插件： ```bash pip install celery-once ``` 配置 Celery： ```python from celery import Celery from celery_once import QueueOnce celery = Celery('tasks', broker='redis://localhost:6379/0') celery.conf.ONCE = { 'backend': 'redis', 'settings': { 'url': 'redis://localhost:6379/0', 'default_timeout': 60 * 5 # 锁的有效期为5分钟 } } class TaskOnce(QueueOnce): once = { 'keys': ['openid'], # 根据 openid 确保任务唯一性 'timeout': 60 * 5, # 锁的有效期为5分钟 'graceful': True # 如果任务已存在，则优雅地忽略 } @celery.task(base=TaskOnce) def async_send_message(openid, content): # 实际发送逻辑 pass ``` 通过上述配置，`async_send_message` 任务将根据 `openid` 参数确保唯一性[^2]。 #### 3. 消息幂等性设计 即使无法完全避免任务重复执行，也可以通过设计幂等性的任务逻辑来减少影响。例如，在数据库中记录任务的状态和结果，确保即使任务多次执行也不会产生副作用。 ```python from sqlalchemy import create_engine, Column, Integer, String, Boolean from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker Base = declarative_base() class MessageTask(Base): __tablename__ = 'message_tasks' id = Column(Integer, primary_key=True) openid = Column(String, unique=True) content = Column(String) is_sent = Column(Boolean, default=False) engine = create_engine('sqlite:///tasks.db') Session = sessionmaker(bind=engine) @celery.task def async_send_message(openid, content): session = Session() task = session.query(MessageTask).filter_by(openid=openid).first() if not task: task = MessageTask(openid=openid, content=content) session.add(task) session.commit() if not task.is_sent: # 实际发送逻辑 task.is_sent = True session.commit() ``` 这种设计确保了即使任务被多次触发，也只会实际执行一次发送操作[^3]。 #### 4. 配置消息队列的幂等性 某些消息队列（如 RabbitMQ 或 Redis）支持消息去重功能。例如，Redis Streams 支持基于消息 ID 的去重机制，而 RabbitMQ 则可以通过设置消息的 `x-message-ttl` 和 `x-dead-letter-exchange` 来实现类似效果。 ```python celery = Celery('tasks', broker='redis://localhost:6379/0') @celery.task def async_send_message(openid, content): # 在任务开始前检查消息是否已处理 if redis_client.exists(f"processed:{openid}"): return # 已处理的消息直接返回 # 实际发送逻辑 redis_client.set(f"processed:{openid}", "true", ex=60 * 5) # 记录已处理的消息 ``` 通过这种方式，可以利用消息队列本身的特性来避免任务重复执行[^4]。 --- ###