RL_Scheduler 项目 Netty + Redis 架构解析#

1
  // 关键配置
2
  EventLoopGroup bossGroup = new NioEventLoopGroup(1);    // 1个Accept线程
3
  EventLoopGroup workerGroup = new NioEventLoopGroup();    // 无限大Worker线程池
4

5
  ServerBootstrap b = new ServerBootstrap();
6
  b.group(bossGroup, workerGroup)
7
      .channel(NioServerSocketChannel.class)
8
      .option(ChannelOption.SO_BACKLOG, 128)      // TCP全连接队列
9
      .childOption(ChannelOption.SO_KEEPALIVE, true)  // TCP保活

1
  ┌─────────────────────┬───────────┬──────────────────┐
2
  │        配置         │    值     │       作用       │
3
  ├─────────────────────┼───────────┼──────────────────┤
4
  │ NioEventLoopGroup() │ 默认CPU*2 │ 高效I/O多路复用  │
5
  ├─────────────────────┼───────────┼──────────────────┤
6
  │ SO_KEEPALIVE        │ true      │ 检测死连接       │
7
  ├─────────────────────┼───────────┼──────────────────┤
8
  │ SO_BACKLOG          │ 128       │ 缓解瞬时连接洪峰 │
9
  └─────────────────────┴───────────┴──────────────────┘

1
  ┌─────────────────────────────────────────┐
2
  │           4B Magic + 1B Version         │
3
  │         + 4B Length + 1B Type          │
4
  ├─────────────────────────────────────────┤
5
  │              Body (Protobuf)            │
6
  └─────────────────────────────────────────┘
7
                Total: 10B Header

1
  // 直接序列化Protobuf，无中间转换
2
  if (msg.getBody() instanceof MessageLite) {
3
      bodyBytes = ((MessageLite) msg.getBody()).toByteArray();
4
  }

1
  ┌───────────────────┬──────┬──────────┬────────┐
2
  │    序列化方式     │ 体积 │ 解析速度 │ 跨语言 │
3
  ├───────────────────┼──────┼──────────┼────────┤
4
  │ Protobuf          │ 最小 │ 最快     │ ✅     │
5
  ├───────────────────┼──────┼──────────┼────────┤
6
  │ JSON              │ 大   │ 中等     │ ✅     │
7
  ├───────────────────┼──────┼──────────┼────────┤
8
  │ Java Serializable │ 大   │ 慢       │ ❌     │
9
  └───────────────────┴──────┴──────────┴────────┘

1
  @Component
2
  public class ChannelManager {
3
      // workerId -> Channel (长连接)
4
      private final Map<String, Channel> workerChannels = new ConcurrentHashMap<>();
5

6
      // 高性能并发读写
7
      public void register(String workerId, Channel channel) {
8
          workerChannels.put(workerId, channel);
9
      }
10

11
      // Master向指定Worker发消息
12
      public void sendToWorker(String workerId, NettyMessage msg) {
13
          Channel ch = workerChannels.get(workerId);
14
          if (ch != null && ch.isActive()) {
15
              ch.writeAndFlush(msg);
16
          }
17
      }
18
  }

1
  // Master收到Worker心跳后：
2
  private void handleHeartbeat(ChannelHandlerContext ctx, HeartbeatRequest req) {
3
      String workerId = req.getWorkerId();
4

5
      // 1. 维护TCP长连接映射
6
      channelManager.register(workerId, ctx.channel());
7

8
      // 2. Redis心跳 (30秒TTL)
9
      String hbKey = "worker:" + workerId + ":hb";
10
      redisTemplate.opsForValue().set(hbKey, "alive", 30, TimeUnit.SECONDS);
11

12
      // 3. 任务运行时续约 (120秒TTL)
13
      if (currentTaskId != null) {
14
          String taskKey = "worker:" + workerId + ":task";
15
          redisTemplate.expire(taskKey, 120, TimeUnit.SECONDS);
16
      }
17
  }

1
Worker: 每10秒发送心跳 ──────────────────────────────▶
2
                                   │
3
Master: 更新Redis TTL ────────────────────────────────▶
4
         worker:{id}:hb  TTL=30s
5
         worker:{id}:task  TTL=120s
6

7
如果Worker宕机：
8
  Redis TTL过期 ──▶ Master检测到"worker超时" ──▶ 重新调度任务

1
  // Worker使用Lettuce同步客户端
2
  RedisURI uri = RedisURI.builder().withHost(host).withPort(port).build();
3
  RedisClient client = RedisClient.create(uri);
4
  StatefulRedisConnection<String, String> connection = client.connect();
5
  RedisCommands<String, String> commands = connection.sync();
6

7
  // 关键方法
8
  public void renew(String currentTaskId, String lastTaskId) {
9
      commands.setex(hbKey(), hbTtlSeconds, "alive");           // 心跳
10
      commands.setex(taskKey(), taskTtlSeconds, currentTaskId); // 任务
11
      commands.setex(taskOwnerKey(currentTaskId), taskTtlSeconds, workerId);
12
  }

1
  ┌─────────┬──────────┬────────┬────────┐
2
  │ 客户端  │ 线程安全 │ 连接池 │ 响应式 │
3
  ├─────────┼──────────┼────────┼────────┤
4
  │ Lettuce │ ✅       │ 内置   │ ✅     │
5
  ├─────────┼──────────┼────────┼────────┤
6
  │ Jedis   │ ⚠️需池   │ 需池   │ ❌     │
7
  └─────────┴──────────┴────────┴────────┘

1
  Master                              Worker
2
     │                                    │
3
     │ EXECUTE_TASK (Protobuf)            │
4
     │───────────────────────────────────▶│ channelRead0()
5
     │                                    │
6
     │  WorkerHandler.runPythonTask()      │
7
     │      │                             │
8
     │      ├── persistTaskStart(taskId)  │
9
     │      │   └── Redis SETEX task key  │
10
     │      │                             │
11
     │      └── 启动Python进程             │
12
     │                                    │
13
     │ EXECUTE_TASK_RESPONSE              │
14
     │◀──────────────────────────────────│ ctx.writeAndFlush()
15
     │                                    │
16
     │         (异步日志流)                │
17
     │◀──────────────────────────────────│ LOG_DATA (每行日志)
18
     │                                    │
19
     │         (任务结束)                  │
20
     │◀──────────────────────────────────│ TASK_STATUS_REPORT
21
     │                                    │
22
     │  releaseTaskOwner(taskId)          │
23
     │  tryDispatchQueuedTaskToWorker()   │

1
  Worker                              Redis                           Master
2
     │                                    │                              │
3
     │──── SETEX worker:1:hb ────────────▶│                              │
4
     │──── SETEX worker:1:task ──────────▶│                              │
5
     │                                    │                              │
6
     │         (10秒后)                    │                              │
7
     │──── SETEX worker:1:hb ────────────▶│                              │
8
     │──── SETEX worker:1:task ──────────▶│                              │
9
     │                                    │                              │
10
     │         (如果宕机)                   │                              │
11
     │     (不再发送心跳)                   │                              │
12
     │                                    │  TTL=30s expire              │
13
     │                                    │◀──── worker:1:hb 消失        │
14
     │                                    │                              │
15
     │                               Master检测到心跳丢失                │
16
     │                               重新调度任务                        │

1
  // 当前配置
2
  .childOption(ChannelOption.SO_KEEPALIVE, true)    // TCP保活
3
  .childOption(ChannelOption.TCP_NODELAY, true)      // 禁用Nagle，低延迟
4
  .childOption(ChannelOption.SO_LINGER, 0)           // 关闭时立即销毁
5

6
  // 可增强配置
7
  .childOption(ChannelOption.SO_RCVBUF, 65536)       // 增大接收缓冲
8
  .childOption(ChannelOption.SO_SNDBUF, 65536)      // 增大发送缓冲

1
  // Pipeline批量操作（减少网络RTT）
2
  List<String> keys = Arrays.asList(
3
      "worker:" + workerId + ":hb",
4
      "worker:" + workerId + ":task"
5
  );
6
  redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
7
      for (String key : keys) {
8
          connection.stringCommands().setEx(
9
              key.getBytes(),
10
              ttlSeconds,
11
              "alive".getBytes()
12
          );
13
      }
14
      return null;
15
  });

1
  ┌─────────────────┬────────────────────────────┬────────────┐
2
  │      当前       │          可优化为          │    收益    │
3
  ├─────────────────┼────────────────────────────┼────────────┤
4
  │ 同步Redis命令   │ Lettuce异步/VirtualThread  │ 更高并发   │
5
  ├─────────────────┼────────────────────────────┼────────────┤
6
  │ Thread-per-Task │ Netty EventLoop            │ 减少线程数 │
7
  ├─────────────────┼────────────────────────────┼────────────┤
8
  │ 单Reactor       │ 多Reactor (boss1, workerN) │ 多核扩展   │
9
  ├─────────────────┼────────────────────────────┼────────────┤
10
  │ 无连接池        │ Lettuce连接池              │ 复用连接   │
11
  └─────────────────┴────────────────────────────┴────────────┘

1
  1. 心跳续约 + Redis TTL = 分布式故障检测
2
     ├── 简单可靠，无需额外监控
3
     └── TTL过期 = Worker宕机/失联
4

5
  2. TaskIdKey 双重保险
6
     ├── worker:{id}:task (Worker自我报告)
7
     └── task:{id}:workerId (调度所有权)
8
     └── 两者不一致 = 任务异常，需要修复
9

10
  3. LogManager 异步解耦
11
     ├── MasterHandler 接收日志后 enqueue
12
     └── 独立线程异步写入文件
13
     └── 不阻塞EventLoop
14

15
  4. ChannelManager 本地缓存
16
     ├── 无需每次查Redis
17
     └── O(1) Channel查找
18
     └── 但需要额外机制处理Worker迁移

1
  ┌──────┬──────────────────────┬───────────────────┐
2
  │ 方面 │       当前实现       │    生产级优化     │
3
  ├──────┼──────────────────────┼───────────────────┤
4
  │ 协议 │ Protobuf ✅          │ 可加压缩 (Snappy) │
5
  ├──────┼──────────────────────┼───────────────────┤
6
  │ 连接 │ 长连接 ✅            │ 连接池化          │
7
  ├──────┼──────────────────────┼───────────────────┤
8
  │ 线程 │ NioEventLoopGroup ✅ │ 可绑核 (Epoll)    │
9
  ├──────┼──────────────────────┼───────────────────┤
10
  │ 心跳 │ Redis续约 ✅         │ 可用Netty内置     │
11
  ├──────┼──────────────────────┼───────────────────┤
12
  │ 调度 │ 简单队列             │ 可加优先级队列    │
13
  ├──────┼──────────────────────┼───────────────────┤
14
  │ 容错 │ 心跳检测 ✅          │ 可加重试+幂等     │
15
  └──────┴──────────────────────┴───────────────────┘