基于Redission实现分布式调度任务选主
在Spring Cloud微服务架构中，基于Redisson实现分布式调度任务的选主和心跳监听，可以通过以下完整方案实现。该方案结合了Redisson的分布式锁和发布/订阅功能，确保高可靠性和实时性：一、整体架构设计
graph TD A[微服务实例1] -->|竞争锁| B(Redisson) C[微服务实例2] -->|竞争锁| B D[微服务实例3] -->|竞争锁| B B -->|心跳事件| E E --> A E --> C E --> D二、核心实现步骤
1. 添加依赖
<dependency>
<groupId>org.redisson</groupId>
redisson-spring-boot-starter</artifactId>
<version>3.23.2</version>
</dependency>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;
@Service
public 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. 健康检查增强
@Service
public 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. 多级故障检测
检测方式触发条件恢复动作Redisson看门狗超时锁续期失败（默认30秒）自动释放锁，其他节点可竞争主动心跳超时自定义阈值（如30秒）强制释放锁并重新选举Redis连接断开ConnectionState.LOST暂停选举直到连接恢复2. 选举性能优化配置
# application.yml
redisson:
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架构选择
部署模式适用场景建议配置哨兵模式高可用要求高3哨兵+3Redis实例Cluster模式大数据量+高性能至少6节点（3主3从）单节点仅开发测试不推荐生产使用2. 监控指标
// 暴露Redisson指标（配合Spring Boot Actuator）
@Bean
public 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
@Primary
public 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);
}
}六、方案优势总结
[*]亚秒级故障检测：通过Redis Pub/Sub实现实时通知
[*]自动故障转移：Redisson看门狗机制保障锁释放
[*]弹性扩展：支持动态增减微服务实例
[*]最小依赖：仅需Redis集群，无需额外组件
[*]与Spring生态无缝集成：完美配合@Scheduled等组件
来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！
页:
[1]