Master-Worker 弹性恢复修复文档#

1
channel.closeFuture().addListener((ChannelFutureListener) closeFuture -> {
2
    LOG.warn(">>> 与 Master 连接断开，5秒后重连...");
3
    closeFuture.channel().eventLoop().schedule(() -> connect(b), 5, TimeUnit.SECONDS);
4
});

1
1. Master 写入 task:{taskId}:workerId → Redis ✓
2
2. Master 发送 ExecuteTaskRequest → Netty
3
3. Worker 接收，设置 currentTaskId，返回确认
4
4. Worker 发送 ExecuteTaskResponse
5
5. Worker 在下次心跳（10秒间隔）时才写入 worker:{workerId}:task → Redis ✗

1
// 原逻辑：只检查 taskOwnerKey 是否存在
2
String workerId = redisTemplate.opsForValue().get(taskWorkerKey(taskId));
3
if (workerId == null || workerId.isBlank()) {
4
    continue;  // 无 owner → 跳过（实际上这是调度中断，应标记 PENDING）
5
}
6
Boolean hbAlive = redisTemplate.hasKey("worker:" + workerId + ":hb");
7
if ((hbAlive != null && hbAlive) && (taskKeyExists != null && taskKeyExists)) {
8
    continue;  // Worker 存活且 taskKey 存在 → 跳过（但无法判断任务是否真的在运行）
9
}

1
Worker: currentTaskId = ""（已清空）
2
Worker: 发送心跳（currentTaskId = ""）
3
Master: 看到空闲 Worker，尝试分配新任务
4
问题: 原任务可能分配给其他 Worker，而原任务仍在 Redis 中有记录

1
T0: Master 写入 taskOwnerKey（Redis）
2
T0+100ms: Netty 消息发送
3
T0+200ms: Worker 接收，设置 currentTaskId
4
T0+10s: Worker 发送心跳（写入 workerTaskKey）
5

6
Gap [T0, T0+10s]: taskOwnerKey 存在，workerTaskKey 不存在
7
→ 修复后 persistTaskStart() 在 T0+205ms（任务接收时）立即写入 workerTaskKey
8
→ Gap 缩小到 ~0ms（原子性）

1
private static final long INITIAL_RECONNECT_DELAY_SECONDS = 1;    // 初始延迟 1 秒
2
private static final long MAX_RECONNECT_DELAY_SECONDS = 60;       // 最大 60 秒
3
private static final double BACKOFF_MULTIPLIER = 2.0;            // 2 倍退避
4
private static final double JITTER_FACTOR = 0.25;                // ±25% 抖动

1
第1次: 1s → 2s → 4s → 8s → 16s → 32s → 60s → 60s → ...

1
private void connect(Bootstrap b) {
2
    if (isReconnecting) {
3
        return;  // 防止并发重连
4
    }
5
    isReconnecting = true;
6
    b.connect(masterHost, masterPort).addListener((ChannelFutureListener) future -> {
7
        if (future.isSuccess()) {
8
            currentReconnectDelay = INITIAL_RECONNECT_DELAY_SECONDS;  // 成功后重置
9
            isReconnecting = false;
10
            // ... 连接成功处理
11
        } else {
12
            scheduleReconnect(b);  // 指数退避重连
13
        }
14
    });
15
}
16

17
private void scheduleReconnect(Bootstrap b) {
18
    long delayWithJitter = calculateDelayWithJitter(currentReconnectDelay);
19
    currentReconnectDelay = Math.min(
20
            (long) (currentReconnectDelay * BACKOFF_MULTIPLIER),
21
            MAX_RECONNECT_DELAY_SECONDS
22
    );
23
    channel.eventLoop().schedule(() -> {
24
        isReconnecting = false;
25
        connect(b);
26
    }, delayWithJitter, TimeUnit.SECONDS);
27
}

1
Worker: connect() → SYN
2
Master: TCP 监听就绪，SYN-ACK 返回
3
Worker: isSuccess() = true → 注册 closeFuture
4
        → channel = future.channel()
5
        → isReconnecting = false
6
Master: Spring 上下文初始化中，Netty pipeline 未就绪
7
        → 心跳无法处理或 NPE
8
        → TCP RST 发送给 Worker
9
Worker: closeFuture 触发 → scheduleReconnect(currentReconnectDelay=60)
10
        → 连接被关闭后立即以 60s 延迟重试
11
        → 循环...

1
private void connect(Bootstrap b) {
2
    if (isReconnecting) {
3
        return;
4
    }
5
    isReconnecting = true;
6
    final Channel[] pendingChannel = new Channel[1];  // 数组包装避免 lambda 捕获
7

8
    b.connect(masterHost, masterPort).addListener((ChannelFutureListener) future -> {
9
        if (!future.isSuccess()) {
10
            scheduleReconnect(b, 0);
11
            return;
12
        }
13

14
        // TCP 握手成功，等待 1 秒确认应用层已就绪
15
        pendingChannel[0] = future.channel();
16
        LOG.info(">>> TCP 握手成功，等待应用层就绪...");
17

18
        pendingChannel[0].eventLoop().schedule(() -> {
19
            if (!pendingChannel[0].isActive()) {
20
                // 1 秒内连接失效 → 立即重连，不等待指数退避
21
                LOG.warn(">>> 连接不稳定（1秒内失效），立即重连...");
22
                pendingChannel[0].close();
23
                scheduleReconnect(b, 0);
24
                return;
25
            }
26

27
            // 连接真正稳定，注册 closeFuture
28
            channel = pendingChannel[0];
29
            currentReconnectDelay = INITIAL_RECONNECT_DELAY_SECONDS;
30
            isReconnecting = false;
31
            LOG.info(">>> 成功连接到 Master（连接已稳定）!");
32

33
            channel.closeFuture().addListener((ChannelFutureListener) closeFuture -> {
34
                if (isReconnecting) {
35
                    return; // 防止双重触发
36
                }
37
                scheduleReconnect(b, 0);
38
            });
39
        }, 1, TimeUnit.SECONDS);
40
    });
41
}

1
private void handleExecuteTask(ChannelHandlerContext ctx, ExecuteTaskRequest req) {
2
    String taskId = req.getTaskId();
3
    this.lastTaskId = taskId;
4
    this.currentTaskId = taskId;
5
    this.currentAttempt = req.getAttempt();
6

7
    // ★ 在启动 Python 线程之前立即写入 Redis（原子性）
8
    if (leaseManager != null) {
9
        leaseManager.persistTaskStart(taskId);
10
    }
11

12
    // 1. 返回响应
13
    // 2. 异步执行 Python 脚本
14
    new Thread(() -> runPythonTask(ctx, req), "Task-Executor-" + taskId).start();
15
}

1
private void reportStatus(ChannelHandlerContext ctx, String taskId, String status, String errorMsg) {
2
    // ★ 立即清除 Redis 任务所有权
3
    if (leaseManager != null) {
4
        leaseManager.clearTask(taskId);
5
    }
6
    // 发送状态报告...
7
}

1
public void persistTaskStart(String taskId) {
2
    // 同时写入两个 key，保证双向一致性
3
    commands.setex(taskKey(), taskTtlSeconds, taskId);           // worker:{workerId}:task
4
    commands.setex(taskOwnerKey(taskId), taskTtlSeconds, workerId);  // task:{taskId}:workerId
5
}
6

7
public void clearTask(String taskId) {
8
    commands.del(taskKey());
9
    if (taskId != null && !taskId.isBlank()) {
10
        commands.del(taskOwnerKey(taskId));
11
    }
12
}

1
for (TrainingTask task : running) {
2
    String taskId = task.getId();
3
    String workerId = redisTemplate.opsForValue().get(taskWorkerKey(taskId));
4

5
    if (workerId == null || workerId.isBlank()) {
6
        // ★ 情况1: taskOwnerKey 不存在 → 调度中断，从未真正开始
7
        task.setStatus("PENDING");
8
        taskMapper.updateById(task);
9
        schedulerService.enqueueTask(taskId);
10
        LOG.info(">>> [Recovery] 任务 [{}] 从未分发（无 owner key），标记为 PENDING", taskId);
11
        continue;
12
    }
13

14
    Boolean hbAlive = redisTemplate.hasKey("worker:" + workerId + ":hb");
15
    if (hbAlive != null && hbAlive) {
16
        // ★ 情况2: Worker 存活 → 任务在运行或刚完成（通知可能丢失）
17
        // 保持 RUNNING，由心跳处理器判断
18
        continue;
19
    }
20

21
    // ★ 情况3: Worker 已死 → 任务孤立
22
    task.setStatus("PENDING");
23
    taskMapper.updateById(task);
24
    schedulerService.enqueueTask(taskId);
25
    schedulerService.releaseTaskOwner(taskId);
26
    LOG.info(">>> [Recovery] 任务 [{}] 的 Worker [{}] 已失活，标记为 PENDING", taskId, workerId);
27
}

1
private void handleHeartbeat(ChannelHandlerContext ctx, HeartbeatRequest req) {
2
    String currentTaskId = req.getCurrentTaskId();
3

4
    if (currentTaskId != null && !currentTaskId.isEmpty()) {
5
        // 续期 TTL
6
        schedulerService.renewTaskOwnerTtl(currentTaskId);
7

8
        // ★ 检查 taskOwnerKey 是否匹配（通知丢失检测）
9
        String ownerKey = "task:" + currentTaskId + ":workerId";
10
        String ownerWorkerId = redisTemplate.opsForValue().get(ownerKey);
11
        if (ownerWorkerId == null || !ownerWorkerId.equals(workerId)) {
12
            // taskOwnerKey 缺失 → 任务已完成但通知丢失
13
            TrainingTask task = taskMapper.selectById(currentTaskId);
14
            if (task != null && "RUNNING".equals(task.getStatus())) {
15
                task.setStatus("COMPLETED");
16
                task.setCompletedAt(LocalDateTime.now());
17
                taskMapper.updateById(task);
18
                LOG.info(">>> [Heartbeat] 任务 [{}] 实际已完成（通知丢失），强制标记为 COMPLETED", currentTaskId);
19
                messagingTemplate.convertAndSend("/topic/tasks", task);
20
            }
21
            schedulerService.tryDispatchQueuedTaskToWorker(workerId);
22
        }
23
    } else {
24
        // ★ Worker 空闲时检查并修复孤儿 RUNNING 任务
25
        checkAndFixStaleRunningTasks(workerId);
26
        schedulerService.tryDispatchQueuedTaskToWorker(workerId);
27
    }
28
}

1
@PostConstruct
2
public void reconstructWorkerTasksFromRedis() {
3
    LOG.info(">>> [SchedulerService] 开始从 Redis 重建 Worker 任务状态...");
4

5
    Set<String> workerTaskKeys = redisTemplate.keys("worker:*:task");
6
    for (String workerTaskKey : workerTaskKeys) {
7
        String workerId = workerTaskKey.split(":")[1];
8
        String taskId = redisTemplate.opsForValue().get(workerTaskKey);
9

10
        // 检查 taskOwnerKey 是否匹配
11
        String ownerKey = taskWorkerKey(taskId);
12
        String ownerWorkerId = redisTemplate.opsForValue().get(ownerKey);
13
        if (ownerWorkerId == null || !ownerWorkerId.equals(workerId)) {
14
            // 调度被中断，标记为 PENDING 重新入队
15
            TrainingTask task = taskMapper.selectById(taskId);
16
            if (task != null && "RUNNING".equals(task.getStatus())) {
17
                task.setStatus("PENDING");
18
                taskMapper.updateById(task);
19
                enqueueIfEnabled(taskId);
20
                LOG.info(">>> [Recovery] 启动重建：任务 [{}] 调度中断，标记为 PENDING", taskId);
21
            }
22
        }
23
    }
24
}

1
Worker 接收任务
2
  → persistTaskStart(taskId)
3
    → SETEX worker:{workerId}:task TTL=120s
4
    → SETEX task:{taskId}:workerId TTL=120s
5
  → 启动 Python 训练线程
6
  → (训练中) heartbeat 每 10s 续期 TTL
7
  → (完成) reportStatus()
8
    → clearTask(taskId)
9
      → DEL worker:{workerId}:task
10
      → DEL task:{taskId}:workerId
11
    → 发送 TASK_STATUS_REPORT

1
Master 重启
2
  → @PostConstruct reconstructWorkerTasksFromRedis()
3
    → 扫描 worker:*:task
4
    → 检查 taskOwnerKey 是否匹配
5
    → 不匹配 → 标记 PENDING，重新入队
6

7
同时 RunningTaskRecovery 每 5s 扫描
8
  → taskOwnerKey 不存在 → 标记 PENDING
9
  → Worker hb 不存在 → 标记 PENDING，释放 owner
10

11
Worker 重连
12
  → 指数退避（1s → 2s → 4s → ...）
13
  → 发送心跳（携带 currentTaskId）
14
  → Master 检查 taskOwnerKey 匹配性
15
  → 匹配 → 任务正常运行
16
  → 不匹配 → 标记 COMPLETED（通知丢失）

1
@Test
2
void testExponentialBackoffWithJitter() {
3
    WorkerAgent agent = new WorkerAgent("localhost", 9000, "test-worker");
4
    long baseDelay = 1;
5

6
    // 计算 5 次延迟，验证指数增长
7
    long delay1 = calculateDelayWithJitter(1);
8
    long delay2 = calculateDelayWithJitter(2);
9
    long delay3 = calculateDelayWithJitter(4);
10
    long delay4 = calculateDelayWithJitter(8);
11

12
    // 抖动范围：delay * (1 - JITTER_FACTOR) <= actual <= delay * (1 + JITTER_FACTOR)
13
    assertTrue(delay1 >= 0.75 && delay1 <= 1.25);
14
    assertTrue(delay2 >= 1.5 && delay2 <= 2.5);
15
    assertTrue(delay3 >= 3.0 && delay3 <= 5.0);
16

17
    // 最大延迟不超过 MAX_RECONNECT_DELAY_SECONDS
18
    long delay60 = calculateDelayWithJitter(60);
19
    assertTrue(delay60 <= 75); // 60 * 1.25
20
}

1
@Test
2
void testPersistTaskStartWritesBothKeys() {
3
    RedisLeaseManager manager = new RedisLeaseManager("worker-1");
4

5
    manager.persistTaskStart("task-123");
6

7
    // 验证两个 key 都存在
8
    assertEquals("task-123", redis.get("worker:worker-1:task"));
9
    assertEquals("worker-1", redis.get("task:task-123:workerId"));
10

11
    manager.clearTask("task-123");
12

13
    // 验证两个 key 都被删除
14
    assertNull(redis.get("worker:worker-1:task"));
15
    assertNull(redis.get("task:task-123:workerId"));
16
}

1
@Test
2
void testTaskRecoversAfterMasterRestart() {
3
    // 1. 提交任务，Worker 接收并开始训练
4
    String taskId = submitTask();
5
    worker.receiveTask(taskId);
6
    assertEquals("RUNNING", db.getTaskStatus(taskId));
7

8
    // 2. Master 宕机（模拟）
9
    master.stop();
10

11
    // 3. Worker 继续训练（currentTaskId 不变）
12
    assertEquals(taskId, worker.getCurrentTaskId());
13
    assertTrue(redis.exists("task:" + taskId + ":workerId"));
14
    assertTrue(redis.exists("worker:" + worker.getId() + ":task"));
15

16
    // 4. Master 重启
17
    master.restart();
18

19
    // 5. Worker 重连（指数退避后成功）
20
    worker.reconnectEventually();
21

22
    // 6. 验证任务仍被追踪为 RUNNING
23
    assertEquals("RUNNING", db.getTaskStatus(taskId));
24
    // taskOwnerKey 存在且匹配，任务保持 RUNNING
25
}

1
@Test
2
void testTaskRequeuedWhenDispatchInterrupted() {
3
    // 1. Master 写入 taskOwnerKey（Redis）
4
    redis.set("task:task-X:workerId", "worker-1");
5

6
    // 2. Master 宕机（尚未发送 Netty 消息）
7
    master.stop();
8

9
    // 3. RunningTaskRecovery 扫描
10
    recovery.recoverOrphanRunningTasks();
11

12
    // 4. 验证：taskOwnerKey 存在但 taskKey 不存在
13
    // → 任务从未分发，标记为 PENDING
14
    assertEquals("PENDING", db.getTaskStatus("task-X"));
15
    assertTrue(queue.contains("task-X"));
16
}

1
@Test
2
void testTaskMarkedCompletedWhenNotificationLost() {
3
    // 1. Worker 完成任务，清除 Redis key
4
    worker.clearTask("task-Y");
5
    // 此时：DB=RUNNING, Redis taskOwnerKey=不存在
6

7
    // 2. Master 宕机（未收到 TASK_STATUS_REPORT）
8
    master.stop();
9

10
    // 3. Worker 发送心跳（currentTaskId = ""，已清空）
11
    worker.sendHeartbeat(currentTaskId: "");
12

13
    // 4. Master 重启，Worker 重连
14
    master.restart();
15
    worker.reconnect();
16

17
    // 5. RunningTaskRecovery 扫描
18
    // → taskOwnerKey 不存在 → 标记 PENDING
19

20
    // 6. Worker 空闲，Master 分发新任务
21
    // （旧任务标记 PENDING 后可被重新调度）
22
}

1
@Test
2
void testNoThunderingHerdOnMasterRestart() {
3
    // 1. 启动 10 个 Worker 连接 Master
4
    List<WorkerAgent> workers = startWorkers(10);
5
    workers.forEach(w -> assertTrue(w.isConnected()));
6

7
    // 2. Master 宕机
8
    master.stop();
9

10
    // 3. 记录每个 Worker 首次重连时间
11
    Map<WorkerAgent, Long> firstReconnectTimes = new HashMap<>();
12
    workers.forEach(w -> {
13
        w.onReconnectAttempt(() -> firstReconnectTimes.put(w, System.currentTimeMillis()));
14
    });
15

16
    // 4. Master 重启
17
    master.restart();
18

19
    // 5. 验证：10 个 Worker 的首次重连时间分散在 1s~4s 范围内
20
    // （指数退避 + 抖动，而非全部同时 1s）
21
    long minTime = firstReconnectTimes.values().stream().min(Long::compare).get();
22
    long maxTime = firstReconnectTimes.values().stream().max(Long::compare).get();
23
    assertTrue(maxTime - minTime > 500); // 至少有 500ms 分散
24
}

1
io.netty.channel.ChannelPipelineException:
2
  org.sgj.rljobscheduler.worker.netty.WorkerHandler is not a @Sharable handler,
3
  so can't be added or removed multiple times.