powerjob-remote/powerjob-remote-impl-mu/DESIGN.md
PowerJob 是一个分布式任务调度框架，采用 Server-Worker 架构。在实际部署中，经常遇到以下网络环境限制：
PowerJob Mu Protocol Architecture
┌──────────────────────────────────────────────────────────────────┐
│ Application Layer │
├──────────────────────────────────────────────────────────────────┤
│ Actor System (Handler Registration & Message Routing) │
├──────────────────────────────────────────────────────────────────┤
│ Transport Layer │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │MuTransporter│ │ChannelMgr │ │ConnectionMgr│ │
│ │ │ │ │ │ │ │
│ │- tell() │ │- Worker Reg │ │- Lazy Conn │ │
│ │- ask() │ │- Ask/Resp │ │- Conn Pool │ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
├──────────────────────────────────────────────────────────────────┤
│ Protocol Layer │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │ MuMessage │ │MuMessageCodec│ │Message Types│ │
│ │ │ │ │ │ │ │
│ │- Type │ │- Encode │ │- TELL │ │
│ │- RequestId │ │- Decode │ │- ASK │ │
│ │- Path │ │- Length │ │- RESPONSE │ │
│ │- Payload │ │- JSON │ │- HEARTBEAT │ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
├──────────────────────────────────────────────────────────────────┤
│ Network Layer │
│ Netty Framework │
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│ │EventLoopGroup│ │ Bootstrap │ │ChannelPipe │ │
│ │ │ │ │ │ │ │
│ │- Boss │ │- Server │ │- Codec │ │
│ │- Worker │ │- Client │ │- Handler │ │
│ └─────────────┘ └─────────────┘ └─────────────┘ │
└──────────────────────────────────────────────────────────────────┘
sequenceDiagram
participant W as Worker
participant S as Server
W->>S: 1. 建立TCP连接
W->>S: 2. 发送HEARTBEAT消息 (包含Worker地址)
S->>S: 3. 注册Worker连接到ChannelManager
S->>S: 4. 建立 workerAddress -> Channel 映射
Note over W,S: 连接建立完成，可以双向通讯
W->>S: 5. 定期发送HEARTBEAT (保活)
S->>W: 6. 可以通过已注册连接发送消息
sequenceDiagram
participant W1 as Worker1
participant W2 as Worker2
W1->>W1: 1. 需要向Worker2发送消息
W1->>W1: 2. 检查连接池，无现有连接
W1->>W2: 3. 建立新的TCP连接
W1->>W2: 4. 发送消息
W1->>W1: 5. 连接加入连接池，供后续复用
public class MuCSInitializer implements CSInitializer {
// 根据节点类型初始化不同服务
public void init(CSInitializerConfig config) {
if (config.getServerType() == ServerType.SERVER) {
initServer(); // 启动Netty服务端 + 连接管理器
} else {
initWorker(); // 启动Netty服务端 + 连接管理器
}
}
private void initServer() {
// 1. 创建ServerHandler
// 2. 启动Netty服务端监听
// 3. 初始化连接管理器
}
private void initWorker() {
// 1. 创建WorkerHandler
// 2. 启动Netty服务端监听 (支持Worker间通讯)
// 3. 初始化连接管理器
}
}
public class MuTransporter implements Transporter {
public void tell(URL url, PowerSerializable request) {
if (当前节点是Worker) {
// 使用连接管理器建立到目标的连接
connectionManager.getOrCreateConnection(url.getAddress())
.thenAccept(channel -> channel.writeAndFlush(message));
} else {
// Server端需要区分目标类型
if (目标是Worker) {
// 使用已注册的Worker连接
Channel channel = channelManager.getWorkerChannel(url.getAddress());
channel.writeAndFlush(message);
} else {
// 使用连接管理器连接到目标Server
connectionManager.getOrCreateConnection(url.getAddress())
.thenAccept(channel -> channel.writeAndFlush(message));
}
}
}
public <T> CompletionStage<T> ask(URL url, PowerSerializable request, Class<T> clz) {
// 1. 生成唯一的requestId
// 2. 注册Future到ChannelManager
// 3. 按照tell的逻辑发送ASK消息
// 4. 返回Future等待响应
}
}
public class ChannelManager {
// Worker地址到连接的映射 (Server端用)
private final ConcurrentMap<String, Channel> workerChannels;
// 请求ID到Future的映射 (Ask模式用)
private final ConcurrentMap<String, CompletableFuture<Object>> pendingRequests;
// 请求ID到响应类型的映射 (类型转换用)
private final ConcurrentMap<String, Class<?>> requestResponseTypes;
public void registerWorkerChannel(Address workerAddress, Channel channel) {
String key = workerAddress.getHost() + ":" + workerAddress.getPort();
workerChannels.put(key, channel);
// 监听连接关闭，自动清理映射
}
public void completePendingRequest(String requestId, Object response) {
CompletableFuture<Object> future = pendingRequests.remove(requestId);
Class<?> responseType = requestResponseTypes.remove(requestId);
// 类型转换，解决LinkedHashMap问题
Object convertedResponse = convertResponse(response, responseType);
future.complete(convertedResponse);
}
}
public class MuConnectionManager {
// 目标地址到连接的映射
private final ConcurrentMap<String, Channel> connections;
// 正在建立的连接
private final ConcurrentMap<String, CompletableFuture<Channel>> pendingConnections;
public CompletableFuture<Channel> getOrCreateConnection(Address targetAddress) {
String key = targetAddress.getHost() + ":" + targetAddress.getPort();
// 1. 检查现有连接
Channel existingChannel = connections.get(key);
if (existingChannel != null && existingChannel.isActive()) {
return CompletableFuture.completedFuture(existingChannel);
}
// 2. 检查正在建立的连接
CompletableFuture<Channel> pendingConnection = pendingConnections.get(key);
if (pendingConnection != null) {
return pendingConnection;
}
// 3. 建立新连接
return createNewConnection(targetAddress);
}
}
public class MuMessage implements PowerSerializable {
private MessageType messageType; // 消息类型
private String requestId; // 请求ID (Ask模式)
private String path; // 处理器路径
private Address senderAddress; // 发送方地址 (注册用)
private Object payload; // 消息载荷
private String errorMessage; // 错误信息
}
public enum MessageType {
TELL, // 单向消息
ASK, // 请求消息
RESPONSE, // 响应消息
ERROR, // 错误响应
HEARTBEAT // 心跳消息
}
public class MuMessageCodec extends ByteToMessageCodec<MuMessage> {
@Override
protected void encode(ChannelHandlerContext ctx, MuMessage msg, ByteBuf out) {
byte[] data = OBJECT_MAPPER.writeValueAsBytes(msg);
out.writeInt(data.length); // 长度前缀
out.writeBytes(data); // 消息内容
}
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
if (in.readableBytes() < 4) return; // 长度不足
int length = in.readInt();
if (in.readableBytes() < length) {
in.resetReaderIndex(); // 重置读取位置
return;
}
byte[] data = new byte[length];
in.readBytes(data);
MuMessage message = OBJECT_MAPPER.readValue(data, MuMessage.class);
out.add(message);
}
}
在单向网络环境中，Server 无法主动连接到 Worker，但需要向 Worker 发送消息。
// Worker端：发送心跳注册
MuMessage heartbeat = new MuMessage(
MessageType.HEARTBEAT, null, null,
workerAddress, // 关键：携带Worker地址
null, null
);
channel.writeAndFlush(heartbeat);
// Server端：处理心跳并注册
public void handleHeartbeat(ChannelHandlerContext ctx, MuMessage msg) {
if (msg.getSenderAddress() != null) {
channelManager.registerWorkerChannel(msg.getSenderAddress(), ctx.channel());
}
}
节点启动时立即建立所有连接会导致：
public CompletableFuture<Channel> getOrCreateConnection(Address targetAddress) {
// 1. 检查缓存连接
Channel cached = connectionCache.get(addressKey);
if (cached != null && cached.isActive()) {
return CompletableFuture.completedFuture(cached);
}
// 2. 检查正在建立的连接，避免重复建立
CompletableFuture<Channel> pending = pendingConnections.get(addressKey);
if (pending != null) {
return pending;
}
// 3. 建立新连接
return createNewConnection(targetAddress);
}
不同的通讯场景需要使用不同的连接方式，需要智能路由。
基于调用方类型和目标类型的二维路由表：
public void routeMessage(URL url, MuMessage message) {
if (currentNodeType == ServerType.WORKER) {
// Worker作为发送方：统一使用连接管理器
connectionManager.getOrCreateConnection(url.getAddress())
.thenAccept(channel -> channel.writeAndFlush(message));
} else {
// Server作为发送方：根据目标类型选择策略
if (url.getServerType() == ServerType.WORKER) {
// 目标是Worker：使用已注册的连接
Channel workerChannel = channelManager.getWorkerChannel(url.getAddress());
workerChannel.writeAndFlush(message);
} else {
// 目标是Server：使用连接管理器
connectionManager.getOrCreateConnection(url.getAddress())
.thenAccept(channel -> channel.writeAndFlush(message));
}
}
}
Jackson 反序列化时可能将对象反序列化为 LinkedHashMap，导致类型转换异常。
// 发送Ask请求时记录类型
public <T> CompletionStage<T> ask(URL url, PowerSerializable request, Class<T> clz) {
String requestId = UUID.randomUUID().toString();
channelManager.registerPendingRequest(requestId, future, clz); // 记录期望类型
// ... 发送消息
}
// 接收响应时转换类型
public void completePendingRequest(String requestId, Object response) {
Class<?> expectedType = requestResponseTypes.remove(requestId);
Object convertedResponse = JsonUtils.toJavaObject(response, expectedType);
future.complete(convertedResponse);
}
// 连接断开监听
channel.closeFuture().addListener(future -> {
// 清理连接映射
connectionCache.remove(addressKey);
workerChannels.remove(workerKey);
// 失败所有待处理的请求
failPendingRequests(channel);
});
// 自动重连
public CompletableFuture<Channel> reconnect(Address address) {
return CompletableFuture
.runAsync(() -> Thread.sleep(retryInterval))
.thenCompose(v -> createConnection(address));
}
// Ask请求超时 (JDK8兼容方式)
ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
scheduler.schedule(() -> {
if (!future.isDone()) {
channelManager.removePendingRequest(requestId);
future.completeExceptionally(new TimeoutException("Request timeout"));
}
}, timeout, TimeUnit.SECONDS);
// 连接事件
log.info("[MuConnectionManager] Connected to {}", targetAddress);
log.warn("[MuConnectionManager] Connection failed to {}", targetAddress);
// 消息事件
log.debug("[MuTransporter] Sent {} message to {}", messageType, url);
log.error("[MuHandler] Failed to process message", exception);
// Worker到Server通讯测试
@Test
public void testWorkerToServerCommunication() {
// 1. 启动Server
// 2. 启动Worker并连接到Server
// 3. Worker发送消息到Server
// 4. 验证Server收到消息
}
// 网络故障恢复测试
@Test
public void testNetworkFailureRecovery() {
// 1. 建立正常连接
// 2. 模拟网络中断
// 3. 恢复网络连接
// 4. 验证通讯自动恢复
}
# Server配置
server:
host: 0.0.0.0 # 监听所有网卡
port: 7700 # 监听端口
external_host: 公网IP # 外部访问地址
# Worker配置
worker:
host: 0.0.0.0 # 本地监听地址
port: 27777 # 本地监听端口
server_host: 公网IP # Server地址
server_port: 7700 # Server端口
# Server端：开放监听端口
iptables -A INPUT -p tcp --dport 7700 -j ACCEPT
# Worker端：确保可以访问Server端口
# 通常不需要额外配置，确保出站不受限即可
PowerJob Mu 协议通过创新的连接复用和延迟连接机制，成功解决了单向网络环境下的双向通讯问题。主要技术特点包括：
该协议特别适合云原生、容器化部署以及混合云环境，为PowerJob在复杂网络环境下提供了可靠的通讯基础。
PowerJob Mu Protocol Technical Design
Version: 1.0
Date: January 2025