Spring-Boot-实现一个简单的分布式定时任务(应用篇)
Spring-Boot-实现一个简单的分布式定时任务(应用篇)
一、背景与问题
在微服务架构中，定时任务的分布式执行是常见需求。传统的单体应用中，Spring的@Scheduled注解可以方便地配置定时任务，但随着系统拆分为多个微服务，这种方案存在致命缺陷：
- 任务重复执行：同一任务可能在多个微服务实例中同时执行
- 任务丢失：服务实例异常时可能导致任务未被触发
- 负载不均：任务集中在少数实例上执行
例如，一个订单清理任务，若部署在三个微服务实例上，可能导致三个实例同时执行清理操作，造成数据不一致。而传统的单体应用只能保证一个实例执行任务。
二、基本原理
分布式定时任务的核心是任务协调机制，需要解决三个关键问题：
- 任务分配：确定哪个实例执行任务
- 任务执行：确保任务逻辑安全执行
- 任务恢复：服务实例异常时能恢复任务执行
典型的解决方案是结合分布式锁和任务分片技术。具体实现流程如下：
- 任务调度器获取分布式锁
- 确定需要执行的任务分片
- 执行任务逻辑
- 释放分布式锁
- 处理任务执行异常和重试机制
三、环境准备
我们使用Spring Boot 3.1.5 + Redis 7.0.5实现分布式定时任务。需要准备的环境：
# Redis服务
redis-server --port 6379
# 项目依赖
dependencies {
implementation 'org.springframework.boot:spring-boot-starter'
implementation 'org.springframework.boot:spring-boot-starter-web'
implementation 'org.springframework.boot:spring-boot-starter-data-redis'
implementation 'io.github.resilience4j:resilience4j-circuitbreaker:1.7.3'
implementation 'io.github.resilience4j:resilience4j-rate-limiter:1.7.3'
}四、核心实现
1. 分布式锁实现
@Configuration
public class RedisLockConfig {
@Autowired
private RedisTemplate<String, Object> redisTemplate;
private final String LOCK_KEY = "distributed_task_lock";
public boolean tryLock(String taskId, long expireTime) {
String lockValue = UUID.randomUUID().toString();
try {
// 使用Lua脚本保证原子性
String script = "if redis.call('setnx', KEYS[1],ARGV[1]) == 1 then " +
"redis.call('expire', KEYS[1], ARGV[2]) " +
"return 1 end return 0";
return (Long) redisTemplate.execute(
RedisScript.of(script, String.class), Arrays.asList(LOCK_KEY), lockValue, String.valueOf(expireTime)) == 1;
} catch (Exception e) {
log.error("获取分布式锁异常", e);
return false;
}
}
public void releaseLock(String taskId) {
String script = "if redis.call('get', KEYS[1]) == ARGV[1] then " +
"redis.call('del', KEYS[1]) " +
"return 1 end return 0";
try {
redisTemplate.execute(
RedisScript.of(script, String.class), Arrays.asList(LOCK_KEY), taskId);
} catch (Exception e) {
log.error("释放分布式锁异常", e);
}
}
}关键点解释：
- 使用Lua脚本确保获取锁和设置过期时间的原子性
- 锁值使用UUID避免冲突
- 设置合理过期时间（建议30秒）
- 释放锁时需要校验锁值有效性
2. 任务分片策略
@Component
public class TaskSharder {
private final int MAX_SHARD = 10;
public int getShardIndex(String taskId) {
// 简单的哈希分片策略
return Math.abs(taskId.hashCode() % MAX_SHARD);
}
public List<String> getShardIds(String taskId) {
List<String> shardIds = new ArrayList<>();
for (int i = 0; i < MAX_SHARD; i++) {
shardIds.add("shard_" + i);
}
return shardIds;
}
}3. 任务执行器
@Service
public class TaskExecutor {
@Autowired
private RedisLockConfig redisLockConfig;
@Autowired
private TaskSharder taskSharder;
public void executeTask(String taskId, String taskType) {
if (redisLockConfig.tryLock(taskId, 30_000)) {
try {
List<String> shardIds = taskSharder.getShardIds(taskId);
// 执行具体任务逻辑
for (String shardId : shardIds) {
processShard(taskId, shardId, taskType);
}
} finally {
redisLockConfig.releaseLock(taskId);
}
}
}
private void processShard(String taskId, String shardId, String taskType) {
// 模拟任务处理逻辑
System.out.println("Processing task: " + taskId + " shard: " + shardId + " type: " + taskType);
// 实际业务逻辑应在此处实现
}
}五、完整案例
1. 项目结构
src
├── main
│ ├── java
│ │ └── com.example
│ │ ├── config
│ │ │ └── RedisLockConfig.java
│ │ ├── service
│ │ │ ├── TaskExecutor.java
│ │ │ └── TaskSharder.java
│ │ ├── controller
│ │ │ └── TaskController.java
│ │ └── TaskApplication.java
│ └── resources
│ └── application.yml2. 配置文件
spring:
redis:
host: localhost
port: 6379
password:
lettuce:
pool:
max-active: 8
max-idle: 8
min-idle: 2
max-wait: 10000ms3. 任务控制器
@RestController
public class TaskController {
@Autowired
private TaskExecutor taskExecutor;
@PostMapping("/execute")
public ResponseEntity<String> executeTask(@RequestParam String taskId, @RequestParam String type) {
taskExecutor.executeTask(taskId, type);
return ResponseEntity.ok("任务执行请求已接收");
}
}4. 启动类
@SpringBootApplication
public class TaskApplication {
public static void main(String[] args) {
SpringApplication.run(TaskApplication.class, args);
}
}六、源码解析
1. 分布式锁获取逻辑
String script = "if redis.call('setnx', KEYS[1],ARGV[1]) == 1 then " +
"redis.call('expire', KEYS[1], ARGV[2]) " +
"return 1 end return 0";setnx命令用于设置键值，仅当键不存在时才设置成功expire命令设置键的过期时间- 使用Lua脚本保证这两个操作的原子性
- 如果返回1表示成功获取锁，否则失败
2. 任务分片策略
int shardIndex = Math.abs(taskId.hashCode() % MAX_SHARD);- 使用任务ID的哈希值进行分片
- 可根据业务需求替换为其他分片策略
- 建议分片数与集群节点数保持一致
3. 异常处理机制
try {
// 业务逻辑
} catch (Exception e) {
log.error("任务执行异常", e);
// 可添加重试机制
}- 需要添加重试机制处理任务执行失败的情况
- 可使用Resilience4j的重试组件实现
七、进阶使用
1. 增加任务分片粒度控制
public int getShardIndex(String taskId, int shardCount) {
return Math.abs(taskId.hashCode() % shardCount);
}- 可根据实际节点数动态调整分片数量
- 建议在启动时读取集群节点数进行计算
2. 引入任务分片状态管理
public class TaskShardState {
private String taskId;
private String shardId;
private boolean isProcessing;
private long lastProcessedTime;
// getters and setters
}- 记录每个分片的处理状态
- 用于故障转移和任务重试
3. 结合消息队列实现任务解耦
@RabbitListener(queues = "task_queue")
public void handleTaskMessage(String message) {
TaskMessage taskMessage = JSON.parseObject(message, TaskMessage.class);
taskExecutor.executeTask(taskMessage.getTaskId(), taskMessage.getType());
}- 将任务触发逻辑与执行逻辑解耦
- 提高系统可维护性
八、性能与工程实践
1. 性能优化策略
| 优化点 | 解决方案 | 效果 |
|---|---|---|
| 锁粒度 | 细粒度锁 | 提高并发性 |
| 锁过期时间 | 设置合理值 | 避免死锁 |
| 任务分片 | 均衡分片 | 提高资源利用率 |
| 缓存预热 | 任务预热 | 减少首次执行延迟 |
2. 异常处理机制
public void handleTaskException(String taskId, Exception e) {
log.error("任务执行异常: {}", taskId, e);
// 记录异常日志
// 暂时保存任务状态
// 可配置重试策略
}3. 安全防护措施
public boolean validateTaskRequest(String taskId, String type) {
// 验证任务类型是否合法
// 验证请求来源是否合法
return true;
}- 增加API网关校验
- 使用JWT验证请求来源
- 记录请求日志进行审计
九、常见问题与踩坑
1. 锁未释放导致资源泄露
public void executeTask(String taskId, String type) {
if (redisLockConfig.tryLock(taskId, 30_000)) {
try {
// 业务逻辑
} catch (Exception e) {
// 忽略异常，导致锁未释放
}
}
}解决方法：使用try-finally确保锁释放
public void executeTask(String taskId, String type) {
boolean locked = false;
try {
locked = redisLockConfig.tryLock(taskId, 30_000);
if (!locked) {
return;
}
// 业务逻辑
} catch (Exception e) {
log.error("任务执行异常", e);
} finally {
if (locked) {
redisLockConfig.releaseLock(taskId);
}
}
}2. 分片策略导致任务不均
int shardIndex = Math.abs(taskId.hashCode() % MAX_SHARD);解决方案：采用一致性哈希算法
int shardIndex = ConsistentHashingUtil.getShardIndex(taskId, MAX_SHARD);3. 网络波动导致锁失效
解决方法：设置合理的锁过期时间，使用锁续期机制
public void renewLock(String taskId) {
String script = "if redis.call('get', KEYS[1]) == ARGV[1] then " +
"redis.call('expire', KEYS[1], ARGV[2]) " +
"return 1 end return 0";
redisTemplate.execute(
RedisScript.of(script, String.class), Arrays.asList(LOCK_KEY), taskId, String.valueOf(30_000));
}十、最佳实践
- 锁粒度控制：建议每个任务单独加锁，避免锁竞争
- 过期时间设置：设置合理的锁过期时间（建议30秒）
- 任务分片策略：根据业务需求选择合适的分片算法
- 异常处理机制：添加重试机制处理任务失败
- 监控系统集成：集成Prometheus监控任务执行状态
- 安全防护措施：增加API网关校验和请求签名
- 日志记录：详细记录任务执行过程，便于故障排查
十一、总结
分布式定时任务的实现需要综合考虑任务协调、锁管理、分片策略等多方面因素。通过结合Redis分布式锁和任务分片策略，可以有效解决传统定时任务在微服务架构中的局限性。在实际应用中，需要根据业务场景选择合适的实现方案，注意处理异常情况和性能优化，确保系统的稳定性和可靠性。对于关键业务场景，建议采用更完善的任务调度框架（如Quartz集群模式），而对于简单的定时需求，本文的实现方案已能满足大部分需求。
评论已关闭