如何仅基于RabbitMQ实现多生产者调度任务协调与单次执行

完全依托RabbitMQ原生队列、消费者特性即可满足需求，不需要引入任何第三方协调、存储组件，核心逻辑分生产端高可用调度、消费端幂等两部分实现：

生产端：基于排他队列实现自动故障转移的调度抢锁 #

RabbitMQ原生的排他队列特性天生具备分布式锁的能力，不需要额外实现选主逻辑：

• 给每个调度任务分配全局唯一固定标识taskId（比如按任务功能命名：order_settlement_daily、user_sync_hourly）
• 所有生产者实例启动后、每次和RabbitMQ重连成功后，都针对每个taskId尝试声明排他锁队列，声明参数固定为：

  // 跨客户端通用参数逻辑
  channel.queueDeclare(
    queue = "scheduler_lock:" + taskId,
    durable = true,
    exclusive = true, // 核心：标记为排他队列
    autoDelete = false,
    arguments = null
  )
  

• 排他队列的原生规则：第一个成功声明该队列的连接（即对应某个生产者实例）独占该队列的所有权，其他连接后续声明同名排他队列会直接抛出访问错误；一旦持有队列的连接断开（对应生产者宕机、网络分区掉线），RabbitMQ会自动删除该排他队列。
• 基于这个规则做调度触发逻辑：
  • 实例声明某taskId的锁队列成功：拿到该任务的调度权，本地调度器正常按周期触发，到点就往业务交换机发对应任务的触发消息
  • 实例声明某taskId的锁队列失败：说明该任务已经被其他存活生产者持有调度权，本地直接跳过该任务的调度触发逻辑，不发消息
• 故障转移逻辑完全自动：持有调度权的生产者宕机后，锁队列自动删除，其余存活生产者会在下次重连/定期重试声明时，第一个成功声明的实例自动拿到调度权，继续触发任务，不存在单点问题。

消费端：基于单活跃消费者实现无分布式存储的幂等消费 #

解决消费端重复执行问题，用RabbitMQ 3.8+版本原生支持的**单活跃消费者（Single Active Consumer）**特性即可：

• 声明任务对应的业务消费队列时，固定加上x-single-active-consumer: true参数
• 该特性的原生规则：同一时间队列只会允许一个注册的消费者实例拉取消息，其余注册的消费者全部处于待命状态；如果当前活跃消费者宕机、断开连接，RabbitMQ会自动按注册顺序切换到下一个存活的消费者承接消费流量，自带故障转移能力。
• 基于这个特性做幂等处理：
  • 生产者发触发消息时，给消息设置全局唯一messageId，规则为taskId + 调度时间戳（比如order_settlement_daily_2024052000），同时消息、队列、交换机全部配置持久化，避免重启丢消息
  • 消费者开启手动ACK模式，拿到消息后先查本地内存缓存（比如用带过期时间的LRU缓存就行，只需要保留最近2个调度周期的messageId），如果该messageId已经处理过，直接ACK丢弃消息不执行任务；如果没处理过，执行完任务逻辑再ACK
  • 因为单活跃消费者保证了同一时间只有一个消费者实例能拿到该队列的消息，幂等校验完全不需要跨实例共享状态，本地内存缓存就足够，不需要引入Redis等第三方存储，也不会出现重复消息分散到不同实例导致去重失效的问题。

关键配置注意事项 #

• 生产者、消费者的RabbitMQ连接都要开启自动重连，重连成功后自动重新执行锁队列声明、消费者注册逻辑
• 消费者本地的去重缓存过期时间设置为调度周期的2倍即可，内存占用极低，不会有性能负担
• 不要给触发消息设置过短的TTL，避免消费者故障恢复期消息被自动丢弃
• 生产环境建议用镜像队列/仲裁队列做队列冗余，避免RabbitMQ单节点故障丢队列、丢消息

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