基于Redission实现分布式调度任务选主

在Spring Cloud微服务架构中，基于Redisson实现分布式调度任务的选主和心跳监听，可以通过以下完整方案实现。该方案结合了Redisson的分布式锁和发布/订阅功能，确保高可靠性和实时性：

一、整体架构设计

二、核心实现步骤

1. 添加依赖

<!-- Redisson Starter -->org.redissonredisson-spring-boot-starter3.23.2

2. 选主服务实现

import org.redisson.api.RLock;import org.redisson.api.RedissonClient;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.stereotype.Service;import javax.annotation.PostConstruct;import javax.annotation.PreDestroy;import java.util.concurrent.TimeUnit;@Servicepublic class LeaderElectionService { private static final String LEADER_LOCK_KEY ="scheduler:leader:lock"; private static final String HEARTBEAT_CHANNEL ="scheduler:heartbeat"; @Autowired private RedissonClient redisson; private volatile boolean isLeader = false; private RLock leaderLock; private Thread heartbeatThread; @PostConstruct public void init() { leaderLock = redisson.getLock(LEADER_LOCK_KEY); startElection(); startHeartbeatListener(); } private void startElection() { // 尝试获取领导权（非阻塞式） new Thread(() -> { while (!Thread.currentThread().isInterrupted()) { try { // 尝试获取锁，锁过期时间30秒 boolean acquired = leaderLock.tryLock(0, 30, TimeUnit.SECONDS); if (acquired) { isLeader = true; System.out.println("当前节点当选为Leader"); startHeartbeatTask(); // 启动心跳任务 break; } Thread.sleep(5000); // 每5秒重试一次 } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } }).start(); } private void startHeartbeatTask() { heartbeatThread = new Thread(() -> { while (isLeader && !Thread.currentThread().isInterrupted()) { try { // 1. 续期锁（看门狗机制会自动处理） // 2. 发布心跳 redisson.getTopic(HEARTBEAT_CHANNEL) .publish(System.currentTimeMillis()); Thread.sleep(10000); // 每10秒发送一次心跳 } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } }); heartbeatThread.start(); } private void startHeartbeatListener() { // 监听Leader心跳 redisson.getTopic(HEARTBEAT_CHANNEL) .addListener(Long.class, (channel, heartbeatTime) -> { System.out.println("收到Leader心跳:" + heartbeatTime); // 可在此更新最后一次心跳时间 }); } @PreDestroy public void shutdown() { if (isLeader && leaderLock.isHeldByCurrentThread()) { leaderLock.unlock(); isLeader = false; if (heartbeatThread != null) { heartbeatThread.interrupt(); } } } public boolean isLeader() { return isLeader; }}

3. 健康检查增强

@Servicepublic class HealthCheckService { @Autowired private RedissonClient redisson; private volatile long lastHeartbeatTime = 0; @PostConstruct public void init() { // 定时检查Leader状态 Executors.newSingleThreadScheduledExecutor() .scheduleAtFixedRate(this::checkLeaderStatus, 0, 5, TimeUnit.SECONDS); } private void checkLeaderStatus() { Long currentTime = redisson.getBucket("scheduler:leader:heartbeat").get(); if (currentTime != null) { lastHeartbeatTime = currentTime; } // 超过30秒未收到心跳认为Leader失效 if (System.currentTimeMillis() - lastHeartbeatTime > 30000) { System.out.println("Leader可能已宕机，触发重新选举"); // 可在此触发主动抢锁逻辑 } }}

三、关键优化点

1. 多级故障检测

2. 选举性能优化配置

# application.ymlredisson: lock: watchdog-timeout: 30000 # 看门狗超时时间(ms) threads: 16 # 事件处理线程数 netty-threads: 32 # Netty工作线程数

3. 脑裂防护方案

// 使用Redisson的MultiLock实现多Redis节点锁RLock lock1 = redissonClient1.getLock(LEADER_LOCK_KEY);RLock lock2 = redissonClient2.getLock(LEADER_LOCK_KEY);RLock multiLock = redisson.getMultiLock(lock1, lock2);boolean acquired = multiLock.tryLock(0, 30, TimeUnit.SECONDS);

四、生产环境部署建议

1. Redis架构选择

2. 监控指标

// 暴露Redisson指标（配合Spring Boot Actuator）@Beanpublic RedissonMetricsBinder redissonMetrics(RedissonClient redisson) { return new RedissonMetricsBinder(redisson);}

监控关键指标：

• redisson.executor.active_threads：活跃线程数
• redisson.pubsub.subscriptions：订阅数量
• redisson.connections.active：活跃连接数

3. 灾备方案

• 双活数据中心：通过RedissonClient配置多区域端点

  Config config = new Config();config.useClusterServers() .addNodeAddress("redis://dc1-node1:6379") .addNodeAddress("redis://dc2-node1:6379");

• 降级策略：本地缓存最后已知状态

  @Bean@Primarypublic LeaderService fallbackLeaderService() { return new FallbackLeaderService(redisLeaderService, localCache);}

五、与Spring Cloud集成

1. 调度任务控制

@Scheduled(fixedRate = 5000)public void scheduledTask() { if (leaderElectionService.isLeader()) { // 只有Leader执行的逻辑 processBatchData(); }}

2. 动态配置更新

@RefreshScope@RestController@RequestMapping("/leader")public class LeaderController { @Value("${election.timeout:30000}") private long electionTimeout; @Autowired private LeaderElectionService electionService; @PostMapping("/timeout") public void updateTimeout(@RequestParam long timeout) { // 动态调整选举超时 electionService.setElectionTimeout(timeout); }}

六、方案优势总结

1. 亚秒级故障检测：通过Redis Pub/Sub实现实时通知
2. 自动故障转移：Redisson看门狗机制保障锁释放
3. 弹性扩展：支持动态增减微服务实例
4. 最小依赖：仅需Redis集群，无需额外组件
5. 与Spring生态无缝集成：完美配合@Scheduled等组件