分布式调度系统 — scheduler-worker执行器详解

一、模块定位与职责

scheduler-worker 是分布式任务调度系统中的执行者，负责接收调度中心下发的任务并执行。如果把调度中心比作快递公司的总部，那么 Worker 就是真正的快递员——他们真正干活。

Worker 的职责可以概括为四件事：

二、启动与连接：Worker 如何上线？

2.1 启动流程

Worker 的启动入口是 SchedulerClient.main()，启动过程如下：

1. 创建 Netty 客户端
└── NioEventLoopGroup：处理 I/O 事件
└── Bootstrap：配置连接参数
└── Pipeline 添加 Handler
├── MessageDecoder：字节 → Message
├── MessageEncoder：Message → 字节
└── ClientHandler：业务处理

2. 连接调度中心（127.0.0.1:8080）
└── 同步等待连接建立

3. 生成 Worker ID
└── "worker-" + System.currentTimeMillis()

4. 发送注册消息（TYPE_REGISTER）
└── Body：WorkerInfo JSON（包含 workerId、host、port 等）

5. 启动心跳线程
└── 每 30 秒发送一次心跳

6. 阻塞等待连接关闭

2.2 心跳线程实现

心跳是 Worker 向调度中心证明自己“还活着”的方式：

private void startHeartbeat() {
Thread heartbeatThread = new Thread(() -> {
while (channel != null && channel.isActive()) {
try {
Thread.sleep(30000); // 30 秒间隔
channel.writeAndFlush(Message.heartbeat());
System.out.println("Send heartbeat");
} catch (InterruptedException e) {
break;
}
}
});
heartbeatThread.setDaemon(true); // 守护线程
heartbeatThread.start();
}

为什么用守护线程？

• 守护线程不会阻止 JVM 退出

• 如果 Worker 主线程因为连接断开而退出，心跳线程会自动结束

• 避免资源泄漏

为什么是 30 秒？

• 调度中心的心跳超时阈值是 60 秒

• 30 秒间隔意味着连续丢失 2 次心跳才会被判定为离线

• 足够容忍网络抖动，又不会浪费太多带宽

三、任务执行入口：ClientHandler

3.1 消息处理流程

ClientHandler 是 Worker 处理所有网络消息的地方：

protected void channelRead0(ChannelHandlerContext ctx, Message msg) {
switch (msg.getType()) {
case TYPE_HEARTBEAT:
// 收到心跳响应，打印日志即可
System.out.println("Heartbeat response received");
break;

case TYPE_REQUEST:
// 1. 解析任务
JobContext job = JsonUtil.fromJson(new String(msg.getBody()), JobContext.class);

// 2. 幂等检查
if (executedTasks.contains(job.getTaskId())) {
// 已执行过，直接返回成功
return ExecutionResult.success(job.getJobId(), job.getTaskId(), "Already executed");
}

// 3. 执行任务
ExecutionResult result = execute(job);

// 4. 记录已执行（幂等）
if (result.getSuccess()) {
executedTasks.add(job.getTaskId());
}

// 5. 返回结果
ctx.writeAndFlush(Message.response(JsonUtil.toJson(result).getBytes()));
break;

case TYPE_CACHE_MIGRATE:
// 接收缓存迁移消息
CacheMigrationMessage migrateMsg = JsonUtil.fromJson(
new String(msg.getBody()), CacheMigrationMessage.class
);
// 预加载热点 Key
for (String key : migrateMsg.getHotKeys()) {
cacheService.get(key, () -> loadFromDB(key));
}
break;
}
}

3.2 幂等检查：为什么需要它？

在分布式环境中，同一个任务可能被调度中心重复下发：

幂等检查的方式：

两层防护保证同一个任务不会被执行两次。

面试可能问：“如果 Worker 重启了，executedTasks 里的数据丢失了怎么办？”

回答：重启后从数据库加载已成功执行的任务 ID，初始化 executedTasks。或者依赖数据库唯一索引作为最后一道防线。

四、任务执行：execute()

4.1 执行逻辑

private ExecutionResult execute(JobContext job) {
try {
// 第一步：分片感知预热
if (job.getPreloadKeys() != null) {
for (String key : job.getPreloadKeys()) {
cacheService.get(key, () -> {
System.out.println("[Preload] Loading from DB: " + key);
return "preloaded_data";
});
}
System.out.println("[Preload] Preloaded " + job.getPreloadKeys().size() + " keys");
}

// 第二步：执行业务逻辑
String userId = job.getParams();
String userData = cacheService.get("user:" + userId, () -> {
System.out.println("[DB] Querying database for user: " + userId);
return "{\\"name\\":\\"User" + userId + "\\",\\"level\\":1}";
});

System.out.println("Executing job: " + job.getJobName() + ", shard: " + job.getShardingItem());
System.out.println("User data from cache: " + userData);

// 模拟耗时操作
Thread.sleep(500);

return ExecutionResult.success(job.getJobId(), job.getTaskId(), "Job executed successfully");
} catch (Exception e) {
return ExecutionResult.failure(job.getJobId(), job.getTaskId(), "Execution failed: " + e.getMessage());
}
}

4.2 执行流程解析

任务到达
│
▼
① 分片感知预热
│ 提前加载 preloadKeys 到 L1 缓存
│ 这样后续业务逻辑执行时，数据已经在本地了
▼
② 业务逻辑执行
│ 从 params 中提取业务参数
│ 通过多级缓存读取数据（L1 → L2 → DB）
▼
③ 返回结果
│ 成功 → ExecutionResult.success()
│ 失败 → ExecutionResult.failure()

4.3 为什么要有预热步骤？

预热步骤是分片感知预热的体现：

• 调度中心下发任务时，会携带 preloadKeys 列表

• 这些 Key 是调度中心根据分片信息预先计算出来的

• Worker 在执行任务前先加载这些 Key，让它们进入 L1 缓存

• 业务逻辑执行时，数据已经在本地了，L1 命中率极高

五、多级缓存：CacheService

5.1 为什么需要多级缓存？

如果所有数据都从数据库读取，性能和并发能力会受限。缓存的本质是用空间换时间。

5.2 读取路径（五层防护）

请求 key
│
▼
① Bloom Filter（防穿透）
│ 判断 key 是否可能存在
│ 不存在 → 直接返回 null（不查任何存储）
│
▼
② Caffeine L1（本地缓存）
│ 命中 → 返回数据
│ 未命中 → 继续
│
▼
③ 互斥锁 + double-check（防击穿）
│ 相同 key 只有一个线程能通过
│ 获取锁后再次检查 L1
│
▼
④ Redis L2（分布式缓存）
│ 命中 → 回填 L1 → 返回数据
│ 未命中 → 继续
│
▼
⑤ DataLoader（数据库）
│ 从 DB 加载数据
│ 写入 Redis（随机 TTL 300~360s）+ L1 + Bloom Filter
│ 如果 DB 无数据 → 缓存空值 "NULL"（TTL 60s）

5.3 缓存穿透防护：布隆过滤器

问题：查询一个不存在的 key，每次请求都穿透到数据库。

解决方案：布隆过滤器 + 空值缓存

// 初始化布隆过滤器
private final BloomFilter<String> bloomFilter = BloomFilter.create(
Funnels.stringFunnel(StandardCharsets.UTF_8),
100000, // 预计插入 10 万条
0.01 // 误判率 1%
);

// 查询时
if (!bloomFilter.mightContain(key)) {
System.out.println("[Cache] Bloom filter: key not exists, skip");
return null; // 直接返回，不查 DB
}

布隆过滤器的特点：

• 判断“不存在”是绝对准确的

• 判断“存在”可能有误判（1% 的概率）

• 误判时，空值缓存 "NULL" 作为兜底

5.4 缓存击穿防护：互斥锁

问题：一个热点 Key 过期后，大量请求同时涌入，全部打到数据库。

解决方案：按 key 加互斥锁 + double-check

ReentrantLock lock = keyLocks.computeIfAbsent(key, k -> new ReentrantLock());
lock.lock();
try {
// double-check：获取锁后再次检查 L1
String value = localCache.getIfPresent(key);
if (value != null) {
return value;
}
// 从 Redis 或 DB 加载
// …
} finally {
lock.unlock();
keyLocks.remove(key);
}

关键点：只有第一个请求会去加载数据，其他请求等待。数据加载完成后，后续请求直接从缓存读取。

5.5 缓存雪崩防护：随机 TTL

问题：大量 Key 同时过期，导致 DB 压力暴增。

解决方案：在基础过期时间上增加随机偏移

int baseExpire = 300; // 基础 300 秒
int randomOffset = random.nextInt(60); // 0~60 秒随机
int expireTime = baseExpire + randomOffset;
jedis.setex(key, expireTime, value);
System.out.println("[Cache] Set with expire: " + expireTime + "s");

不同 Key 的过期时间在 300~360 秒之间均匀分布，不会同时过期。

六、缓存迁移接收

当调度中心检测到某个 Worker 下线时，会发送 TYPE_CACHE_MIGRATE 消息，把该 Worker 的热点 Key 迁移到其他 Worker。

case TYPE_CACHE_MIGRATE:
String migrateJson = new String(msg.getBody());
CacheMigrationMessage migrateMsg = JsonUtil.fromJson(migrateJson, CacheMigrationMessage.class);

System.out.println("[Migration] Received cache migration, keys: " + migrateMsg.getHotKeys().size());

for (String key : migrateMsg.getHotKeys()) {
cacheService.get(key, () -> {
System.out.println("[Migration] Loading key: " + key);
return "{\\"migrated\\":true}";
});
}
System.out.println("[Migration] Preloaded " + migrateMsg.getHotKeys().size() + " keys");
break;

迁移后：新 Worker 的本地缓存已经包含了这些热点 Key，任务执行时不会因为缓存缺失而变慢。

七、缓存配置：CacheConfig

public class CacheConfig {
public static <K, V> Cache<K, V> createLocalCache() {
return Caffeine.newBuilder()
.maximumSize(10000) // 最大 10000 条
.expireAfterWrite(60, TimeUnit.SECONDS) // 60 秒过期
.recordStats() // 记录命中率
.build();
}
}

为什么选择 Caffeine？

Caffeine 是目前 Java 生态中最优秀的本地缓存库。

八、关键设计决策总结

九、常见面试问题

Q1：Worker 本地缓存和 Redis 缓存的数据一致性问题怎么处理？

缓存一致性是分布式系统的经典难题。我采用的是最终一致性策略：

Q2：布隆过滤器的误判率怎么来的？

布隆过滤器使用多个哈希函数。当数据量超过预期时，哈希碰撞概率增加。这就是为什么需要配置 expectedInsertions 和 fpp（误判率）。1% 的误判率意味着 100 个不存在的 Key 中可能有 1 个被误判为存在，然后由空值缓存兜底。

Q3：如果 Worker 在执行任务时宕机了，任务怎么办？

调度中心的心跳检测会感知 Worker 下线，然后在 removeWorker 方法中把该 Worker 的 runningTasks 放入重试队列，最终由其他 Worker 接管。

Q4：Worker 收到了 TYPE_CACHE_MIGRATE 消息，但消息中的 HotKey 数量很多，会不会导致大量内存占用？

每次迁移最多 10 个 Key（每个任务预热 10 个 Key），而且只是预热 Key 的元数据，不是预热整个数据内容。实际内存占用很小。

十、三个模块总结

一句话总结：common 定义“怎么说”，core 决定“谁来做”，worker 负责“怎么做”。