java分布式定時任務(wù)實現(xiàn)細節(jié)
一、分布式鎖的底層實現(xiàn)細節(jié)(以 Redis 為例)
分布式鎖是解決任務(wù)重復(fù)執(zhí)行的核心,需保證原子性、超時釋放和可重入性。以下是生產(chǎn)級 Redis 鎖實現(xiàn):
public class RedisDistributedLock {
private final RedisTemplate<String, String> redisTemplate;
private final String lockKey;
private final String lockValue; // 用于標識鎖持有者(支持可重入)
private final long expireMillis; // 鎖過期時間(避免死鎖)
// 構(gòu)造函數(shù):初始化鎖參數(shù)
public RedisDistributedLock(RedisTemplate<String, String> redisTemplate,
String lockKey, String requestId, long expireMillis) {
this.redisTemplate = redisTemplate;
this.lockKey = lockKey;
this.lockValue = requestId; // 建議使用UUID+線程ID
this.expireMillis = expireMillis;
}
// 嘗試獲取鎖(原子操作)
public boolean tryLock() {
// 使用Redis的SET命令實現(xiàn):NX(不存在則設(shè)置)+ PX(毫秒過期)
return redisTemplate.opsForValue().setIfAbsent(lockKey, lockValue, expireMillis, TimeUnit.MILLISECONDS);
}
// 釋放鎖(需校驗持有者,避免誤釋放)
public boolean unlock() {
// 使用Lua腳本保證刪除操作的原子性
String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
Long result = (Long) redisTemplate.execute(
new DefaultRedisScript<>(script, Long.class),
Collections.singletonList(lockKey),
lockValue
);
return result != null && result > 0;
}
// 帶超時等待的獲取鎖(輪詢重試)
public boolean tryLock(long waitTime, TimeUnit unit) throws InterruptedException {
long timeout = unit.toMillis(waitTime);
long start = System.currentTimeMillis();
while (true) {
if (tryLock()) {
return true;
}
// 等待重試(避免自旋過于頻繁)
long remaining = timeout - (System.currentTimeMillis() - start);
if (remaining <= 0) {
return false; // 超時未獲取到鎖
}
Thread.sleep(Math.min(remaining, 100)); // 最多等待100ms重試
}
}
}關(guān)鍵設(shè)計點:
- 鎖標識(lockValue):用 UUID + 線程 ID 區(qū)分持有者,避免釋放其他節(jié)點的鎖。
- 過期時間:需大于任務(wù)執(zhí)行時間(如任務(wù)耗時 5s,鎖過期設(shè) 10s),防止節(jié)點宕機導(dǎo)致鎖永久持有。
- 續(xù)約機制:若任務(wù)執(zhí)行時間可能超過鎖過期時間,需啟動后臺線程定期續(xù)約(如每 3s 續(xù)期 10s)。
二、任務(wù)調(diào)度核心原理(以 XXL-Job 為例)
1. 調(diào)度中心與執(zhí)行器通信流程
- 執(zhí)行器注冊:執(zhí)行器啟動時通過 HTTP 請求向調(diào)度中心注冊(攜帶 appname、IP、端口)。
- 任務(wù)觸發(fā):調(diào)度中心根據(jù) CRON 表達式計算下次執(zhí)行時間,到達時間后通過線程池觸發(fā)任務(wù),向執(zhí)行器發(fā)送 HTTP 請求(POST 方式)。
- 執(zhí)行反饋:執(zhí)行器執(zhí)行完任務(wù)后,將結(jié)果(成功 / 失敗、日志)同步回調(diào)度中心。
2. 路由策略與負載均衡
XXL-Job 支持多種路由策略,解決任務(wù)在集群節(jié)點的分配問題:
- 第一個節(jié)點:固定選擇集群中第一個在線節(jié)點(適合單節(jié)點執(zhí)行的任務(wù))。
- 輪詢:按順序依次分配給在線節(jié)點(均衡負載)。
- 分片廣播:所有在線節(jié)點同時執(zhí)行,每個節(jié)點處理不同分片(適合大規(guī)模任務(wù))。
分片示例:100 萬條數(shù)據(jù)需批量處理,分為 5 個分片,集群 3 個節(jié)點:
@XxlJob("shardingTask")
public ReturnT<String> shardingHandler(String param) {
// 獲取分片參數(shù)(由調(diào)度中心分配)
ShardingUtil.ShardingVO shardingVO = ShardingUtil.getShardingVo();
int shardIndex = shardingVO.getIndex(); // 當前分片索引(0-4)
int shardTotal = shardingVO.getTotal(); // 總分片數(shù)(5)
// 按分片處理數(shù)據(jù)(如按ID取模:id % shardTotal == shardIndex)
List<Data> dataList = dataService.queryBySharding(shardIndex, shardTotal);
for (Data data : dataList) {
processData(data);
}
return ReturnT.SUCCESS;
}三、高可用設(shè)計(避免單點故障)
1. 調(diào)度中心集群化
- 部署方式:多實例部署(如 2 個節(jié)點),通過 Nginx 負載均衡對外提供服務(wù)。
- 數(shù)據(jù)一致性:依賴 MySQL 主從同步(調(diào)度中心數(shù)據(jù)存儲在 MySQL),確保多實例數(shù)據(jù)一致。
2. 執(zhí)行器故障轉(zhuǎn)移
- 心跳檢測:執(zhí)行器定期向調(diào)度中心發(fā)送心跳(默認 30s 一次),超過 90s 未心跳則標記為離線。
- 任務(wù)轉(zhuǎn)移:若執(zhí)行器離線,調(diào)度中心會將其負責(zé)的任務(wù)分配給其他在線節(jié)點(需任務(wù)支持重執(zhí)行)。
四、監(jiān)控與告警體系
1. 核心監(jiān)控指標
- 任務(wù)維度:執(zhí)行次數(shù)、成功率、平均耗時、最大耗時。
- 節(jié)點維度:CPU 使用率、內(nèi)存占用、任務(wù)并發(fā)數(shù)。
2. 集成 Prometheus 監(jiān)控
// 自定義任務(wù)執(zhí)行指標(使用Micrometer)
@Component
public class TaskMetrics {
private final MeterRegistry meterRegistry;
public TaskMetrics(MeterRegistry meterRegistry) {
this.meterRegistry = meterRegistry;
}
// 記錄任務(wù)執(zhí)行耗時
public void recordTaskDuration(String taskName, long durationMs) {
Timer.builder("task.execution.duration")
.tag("task", taskName)
.register(meterRegistry)
.record(durationMs, TimeUnit.MILLISECONDS);
}
// 記錄任務(wù)失敗次數(shù)
public void incrementFailCount(String taskName) {
Counter.builder("task.execution.fail")
.tag("task", taskName)
.register(meterRegistry)
.increment();
}
}在任務(wù)執(zhí)行中埋點:
@XxlJob("orderTimeoutTask")
public ReturnT<String> orderTimeoutHandler(String param) {
long start = System.currentTimeMillis();
try {
// 任務(wù)邏輯...
metrics.recordTaskDuration("orderTimeoutTask", System.currentTimeMillis() - start);
return ReturnT.SUCCESS;
} catch (Exception e) {
metrics.incrementFailCount("orderTimeoutTask");
return ReturnT.FAIL;
}
}3. 告警配置
通過 Grafana 設(shè)置告警規(guī)則(如任務(wù)失敗率 > 5% 時觸發(fā)告警),并集成釘釘 / 企業(yè)微信機器人:
// 釘釘告警示例
public class DingTalkAlarm {
private final String webhook;
public void sendAlarm(String message) {
HttpHeaders headers = new HttpHeaders();
headers.setContentType(MediaType.APPLICATION_JSON);
Map<String, Object> body = new HashMap<>();
body.put("msgtype", "text");
body.put("text", Map.of("content", "定時任務(wù)告警:" + message));
new RestTemplate().postForObject(webhook, new HttpEntity<>(body, headers), String.class);
}
}五、自定義輕量級方案(無框架依賴)
若場景簡單(如無動態(tài)配置需求),可基于 Redis + 線程池實現(xiàn)極簡方案:
@Component
public class RedisScheduledTask {
@Autowired
private RedisTemplate<String, String> redisTemplate;
@Autowired
private TaskService taskService;
// 初始化定時任務(wù)(每分鐘執(zhí)行一次)
@PostConstruct
public void init() {
ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
executor.scheduleAtFixedRate(this::executeTask, 0, 1, TimeUnit.MINUTES);
}
// 執(zhí)行任務(wù)(加分布式鎖)
private void executeTask() {
String lockKey = "task:order:timeout";
String requestId = UUID.randomUUID().toString();
RedisDistributedLock lock = new RedisDistributedLock(redisTemplate, lockKey, requestId, 60000);
try {
if (lock.tryLock()) {
// 執(zhí)行核心邏輯
taskService.processTimeoutOrders();
} else {
log.info("任務(wù)被其他節(jié)點執(zhí)行,當前節(jié)點跳過");
}
} finally {
lock.unlock(); // 釋放鎖
}
}
}六、避坑指南
- 鎖過期時間設(shè)置:需大于任務(wù)最大執(zhí)行時間(可通過壓測評估),避免任務(wù)未執(zhí)行完鎖已釋放。
- 任務(wù)冪等性:即使加了鎖,仍需保證任務(wù)可重復(fù)執(zhí)行(如使用
UPDATE orders SET status=1 WHERE id=? AND status=0)。 - 線程池隔離:核心任務(wù)與非核心任務(wù)使用獨立線程池(如
Executors.newScheduledThreadPool(5)),避免相互阻塞。 - 日志追蹤:任務(wù)執(zhí)行日志需包含唯一 ID(如訂單號),便于問題排查。
通過以上細節(jié)設(shè)計,可構(gòu)建既高效又可靠的分布式定時任務(wù)系統(tǒng),兼顧性能、可用性和可運維性。實際項目中,建議優(yōu)先選用 XXL-Job 等成熟框架,減少重復(fù)開發(fā);特殊場景下再考慮自定義方案。
到此這篇關(guān)于java分布式定時任務(wù)實現(xiàn)細節(jié)的文章就介紹到這了,更多相關(guān)java分布式定時任務(wù)內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
相關(guān)文章
springboot與springmvc基礎(chǔ)入門講解
本篇文章主要介紹了詳解快速搭建Spring Boot+Spring MVC,小編覺得挺不錯的,現(xiàn)在分享給大家,也給大家做個參考。一起跟隨小編過來看看吧2021-07-07
MyBatisPlus+SpringBoot實現(xiàn)樂觀鎖功能詳細流程
樂觀鎖是針對一些特定問題的解決方案,主要解決丟失更新問題,下面這篇文章主要給大家介紹了關(guān)于MyBatisPlus+SpringBoot實現(xiàn)樂觀鎖功能的相關(guān)資料,文中通過實例代碼介紹的非常詳細,需要的朋友可以參考下2023-03-03
詳解Spring注解@Autowired的實現(xiàn)原理和使用方法
在使用Spring開發(fā)的時候,配置的方式主要有兩種,一種是xml的方式,另外一種是 java config的方式,在使用的過程中,我們使用最多的注解應(yīng)該就是@Autowired注解了,所以本文就給大家講講@Autowired注解是如何使用和實現(xiàn)的,需要的朋友可以參考下2023-07-07
Spring Boot console log 格式自定義方式
這篇文章主要介紹了Spring Boot console log 格式自定義方式,具有很好的參考價值,希望對大家有所幫助。如有錯誤或未考慮完全的地方,望不吝賜教2021-07-07

