contract-manager/docs/task/client_server_tasker_communication_rules.md

# 客户端 Tasker 至 服务器端 Tasker 通信规则与逻辑

本文档总结了 Contract-Manager 项目中客户端 Tasker 与服务器端 Tasker 之间的通信规则、调用逻辑和实现模式，基于对以下文件的分析：
- [`CompanyCustomerEvaluationFormUpdateTask.java`](/server/src/main/java/com/ecep/contract/ds/customer/tasker/CompanyCustomerEvaluationFormUpdateTask.java)
- [`CompanyCustomerEvaluationFormUpdateTask.java`](/client/src/main/java/com/ecep/contract/controller/customer/CompanyCustomerEvaluationFormUpdateTask.java)
- [`CustomerTabSkinFile.java`](/client/src/main/java/com/ecep/contract/controller/customer/CustomerTabSkinFile.java)

## 1. 架构设计原则

项目采用了清晰的客户端-服务器分离架构，任务处理遵循以下原则：
- **客户端轻量级**：负责任务发起、参数传递和结果展示
- **服务器端重量级**：负责实际业务逻辑处理和数据操作
- **WebSocket通信**：使用WebSocket实现客户端与服务器端的任务通信和进度同步

### 1.1 核心通信组件

#### WebSocketClientService

[`WebSocketClientService`](/client/src/main/java/com/ecep/contract/WebSocketClientService.java)类是客户端WebSocket通信的核心服务组件，负责建立、维护与服务器的WebSocket连接，并提供消息发送和接收的功能。主要职责包括：

- **连接管理**：初始化WebSocket连接、处理连接关闭和重连逻辑
- **心跳维护**：定期发送心跳消息保持连接活跃
- **会话管理**：创建和管理WebSocket会话（`WebSocketClientSession`）
- **消息路由**：接收并路由服务器消息到对应的会话处理器
- **服务调用**：提供`invoke`方法调用服务器端服务

#### WebSocketClientSession

[`WebSocketClientSession`](/client/src/main/java/com/ecep/contract/WebSocketClientSession.java)类代表一个特定的WebSocket会话，每个Tasker任务执行时都会创建一个对应的会话实例。主要职责包括：

- **会话标识**：维护唯一的会话ID
- **任务提交**：将Tasker任务提交到服务器端执行
- **消息处理**：处理服务器返回的各类消息（包括进度更新、属性更新、状态变更等）
- **属性同步**：根据服务器消息更新Tasker的属性值

## 2. 类命名与结构规范

### 2.1 命名规则
- 客户端与服务器端的对应Tasker类**必须使用相同的类名**（如示例中的`CompanyCustomerEvaluationFormUpdateTask`）
- 客户端Tasker位于`client`模块的控制器包下
- 服务器端Tasker位于`server`模块的tasker包下

### 2.2 任务名称注册规则
- 所有服务器端Tasker类必须通过配置文件`tasker_mapper.json`进行注册，并由`WebSocketServerTaskManager`类在`afterPropertiesSet`方法中加载
- 注册格式为`"TaskClassName": "fully.qualified.ClassName"`
- 注册的Task名称将用于客户端和服务器端之间的任务识别
- 配置文件`tasker_mapper.json`应位于`src/main/resources`目录下，格式如下：

```json
{
    "taskers": {
        "TaskClassName": "fully.qualified.ClassName",
        // 更多任务映射...
    }
}
```

#### 任务名称注册实现示例
```java
@Override
public void afterPropertiesSet() throws Exception {
    // 从tasker_mapper.json文件中加载任务注册信息
    try {
        Resource resource = resourceLoader.getResource("classpath:tasker_mapper.json");
        try (InputStream inputStream = resource.getInputStream()) {
            JsonNode rootNode = objectMapper.readTree(inputStream);
            JsonNode taskersNode = rootNode.get("taskers");
            if (taskersNode != null && taskersNode.isObject()) {
                Map<String, String> taskMap = new java.util.HashMap<>();
                taskersNode.fields().forEachRemaining(entry -> {
                    taskMap.put(entry.getKey(), entry.getValue().asText());
                });
                taskClzMap = taskMap;
            }
        }
    } catch (Exception e) {
        logger.error("Failed to load tasker_mapper.json", e);
        // 使用默认值作为fallback
        taskClzMap = Map.of();
    }
}
```

### 2.3 接口实现区分
- 客户端Tasker实现[`WebSocketClientTasker`](/client/src/main/java/com/ecep/contract/WebSocketClientTasker.java)接口
- 服务器端Tasker实现[`WebSocketServerTasker`](/server/src/main/java/com/ecep/contract/ds/customer/tasker/WebSocketServerTasker.java)接口

### 2.4 继承关系
- 客户端Tasker继承自[`Tasker<Object>`](/client/src/main/java/com/ecep/contract/task/Tasker.java)基类
- 服务器端Tasker继承自[`Tasker<Object>`](/server/src/main/java/com/ecep/contract/ui/Tasker.java)基类

## 3. 客户端Tasker实现规则

客户端Tasker是任务的发起方，需要遵循以下实现规则：

### 3.1 核心属性
- 通常包含一个可设置的业务对象（如示例中的`@Setter private CompanyCustomerVo customer;`）


### 3.2 核心方法实现
- **getTaskName()**：返回任务名称，通常使用类名
- **updateProgress()**：继承或重写进度更新方法
- **execute()**：调用`callRemoteTask()`方法将任务发送到服务器端，传递必要参数, 参数类型只允许基本类和Vo类对象
- **updateProgress()**：继承或重写进度更新方法，public，用于接收服务器端发送的进度更新消息

### 3.3 示例实现
```java
public class CompanyCustomerEvaluationFormUpdateTask extends Tasker<Object> implements WebSocketClientTasker {
    private static final Logger logger = LoggerFactory.getLogger(CompanyCustomerEvaluationFormUpdateTask.class);

    @Setter
    private CompanyCustomerVo customer;  // 业务对象

    @Override
    public String getTaskName() {
        return getClass().getSimpleName();
    }

    @Override
    protected Object execute(MessageHolder holder) throws Exception {
        updateTitle("客户评估表更新任务");  // 设置任务标题
        // 调用远程任务，传递locale和业务对象ID
        return callRemoteTask(holder, getLocale(), customer.getId());
    }
}
```

## 4. 服务器端Tasker实现规则

服务器端Tasker是任务的实际执行者，需要遵循以下实现规则：

### 4.1 参数接收
- 实现`init(JsonNode argsNode)`方法接收客户端传递的参数
- 从参数中提取业务对象ID并加载完整业务对象

### 4.2 服务获取
- 通过`getCachedBean(Service.class)`方法获取所需的服务实例
- 可以提供辅助方法封装服务获取逻辑

### 4.3 任务执行
- 实现`execute(MessageHolder holder)`方法包含实际业务逻辑
- 使用`holder.info()/error()`等方法记录任务执行状态
- 调用`updateProgress()`方法更新任务进度
- 调用`updateTitle()`方法更新任务标题
- 调用`updateProperty()`方法更新任务属性
- `holder`、`updateProgress()`、`updateTitle()`和`updateProperty()` 等方法通过 webSocket 把数据传输到 client 端的 tasker，client 端的 tasker 收到数据后会调用对应的方法更新任务属性

### 4.4 示例实现
```java
public class CompanyCustomerEvaluationFormUpdateTask extends Tasker<Object> implements WebSocketServerTasker {
    private CompanyCustomer customer;  // 业务对象

    @Override
    public void init(JsonNode argsNode) {
        // 从参数中提取业务对象ID并加载
        int customerId = argsNode.get(0).asInt();
        customer = getCachedBean(CompanyCustomerService.class).findById(customerId);
    }

    // 辅助方法获取服务
    CompanyCustomerFileService getCompanyCustomerFileService() {
        return getCachedBean(CompanyCustomerFileService.class);
    }

    @Override
    protected Object execute(MessageHolder holder) throws Exception {
        // 执行实际业务逻辑
        updateEvaluationForm(holder);
        return null;
    }

    // 具体业务逻辑实现
    public void updateEvaluationForm(MessageHolder holder) {
        // 业务逻辑代码...
        updateProgress(1, 10);  // 更新进度
        // 更多业务逻辑...
    }
}
```

## 5. 客户端Tasker调用流程

在UI控制器中调用客户端Tasker的标准流程如下：

### 5.1 实例化与参数设置
1. 创建客户端Tasker实例
2. 设置必要的业务对象参数

### 5.2 任务执行与监控
1. 使用`UITools.showTaskDialogAndWait()`显示任务对话框
2. 调用`initializeTask()`初始化任务执行环境
3. 通过消费者函数处理任务消息

### 5.3 任务完成处理
1. 任务完成后执行必要的数据刷新操作

### 5.4 示例调用
```java
public void onUpdateEvaluationFormAction(ActionEvent event) {
    // 1. 创建Tasker并设置参数
    CompanyCustomerEvaluationFormUpdateTask task = new CompanyCustomerEvaluationFormUpdateTask();
    task.setCustomer(getCompanyCustomerService().findById(viewModel.getId().get()));
    
    // 2. 显示任务对话框并执行任务
    UITools.showTaskDialogAndWait("更新评价表", task, consumer -> {
        initializeTask(task, "更新评价表", msg -> consumer.accept(Message.info(msg)));
    });
    
    // 3. 任务完成后刷新数据
    loadTableDataSet();
}
```

## 6. 任务进度管理

- 服务器端使用`updateProgress(current, total)`方法更新任务进度
- 进度值通常以0-1000或类似小范围数值表示完成百分比
- 客户端通过WebSocket接收进度更新并显示

## 7. 异常处理机制

- 服务器端使用`MessageHolder.error()`方法记录错误信息
- 客户端通过任务对话框展示错误信息
- 服务器端在关键操作点进行异常捕获和处理

## 9. 数据一致性保障

- 任务完成后客户端通常调用数据刷新方法（如`loadTableDataSet()`）确保UI显示最新数据
- 服务器端负责业务数据的持久化操作

## 10. WebSocketServerTaskManager功能与作用

`WebSocketServerTaskManager`是服务器端管理WebSocket任务的核心组件，负责处理客户端发起的任务请求，创建并执行对应的任务实例，并维护任务执行过程中的双向通信。

### 10.1 核心职责

- **任务注册管理**：从`tasker_mapper.json`文件加载并管理任务名称与对应实现类的映射关系
- **消息处理分发**：处理客户端发送的任务创建请求
- **任务实例化**：根据任务名称动态创建服务器端Tasker实例
- **处理器设置**：为Tasker设置各类回调处理器，实现服务器端到客户端的消息推送
- **异步任务执行**：使用虚拟线程异步执行任务
- **消息通信**：维护任务执行过程中的双向通信，包括进度更新、状态通知等
- **异常处理**：处理任务执行过程中的异常情况并通知客户端

### 10.2 任务注册机制

`WebSocketServerTaskManager`在初始化阶段（通过实现`InitializingBean`接口的`afterPropertiesSet()`方法）从`tasker_mapper.json`文件中加载任务注册信息：

```java
@Override
public void afterPropertiesSet() throws Exception {
    // 从tasker_mapper.json文件中加载任务注册信息
    try {
        Resource resource = resourceLoader.getResource("classpath:tasker_mapper.json");
        try (InputStream inputStream = resource.getInputStream()) {
            JsonNode rootNode = objectMapper.readTree(inputStream);
            JsonNode taskersNode = rootNode.get("taskers");
            if (taskersNode != null && taskersNode.isObject()) {
                Map<String, String> taskMap = new java.util.HashMap<>();
                taskersNode.fields().forEachRemaining(entry -> {
                    taskMap.put(entry.getKey(), entry.getValue().asText());
                });
                taskClzMap = taskMap;
            }
        }
    } catch (Exception e) {
        logger.error("Failed to load tasker_mapper.json", e);
        // 使用默认值作为fallback
        taskClzMap = Map.of();
    }
}
```

#### 10.2 消息处理流程

`WebSocketServerTaskManager`的消息处理流程如下：

1. **接收消息**：通过`onMessage`方法接收来自WebSocketSession的消息
2. **解析会话ID**：从消息中提取会话标识
3. **分发处理**：调用`handleAsSessionCallback`根据消息类型进行分发
4. **异常处理**：捕获处理过程中的异常并发送错误消息

```java
public void onMessage(WebSocketSession session, JsonNode jsonNode) {
    // 处理 sessionId 的消息
    String sessionId = jsonNode.get(WebSocketConstant.SESSION_ID_FIELD_NAME).asText();
    try {
        handleAsSessionCallback(session, sessionId, jsonNode);
    } catch (Exception e) {
        sendError(session, sessionId, e.getMessage());
        logger.warn("处理会话回调失败 (会话ID: {}): {}", sessionId, e.getMessage(), e);
    }
}

private void handleAsSessionCallback(WebSocketSession session, String sessionId, JsonNode jsonNode) {
    if (!jsonNode.has("type")) {
        throw new IllegalArgumentException("缺失 type 参数");
    }
    String type = jsonNode.get("type").asText();
    if (type.equals("createTask")) {
        createTask(session, sessionId, jsonNode);
    }
}

private void sendError(WebSocketSession session, String sessionId, String message) {
    if (session == null || !session.isOpen()) {
        logger.warn("尝试向已关闭的WebSocket会话发送错误消息: {}", message);
        return;
    }

    try {
        String errorMessage = objectMapper.writeValueAsString(Map.of(
                WebSocketConstant.SESSION_ID_FIELD_NAME, sessionId,
                WebSocketConstant.SUCCESS_FIELD_VALUE, false,
                WebSocketConstant.MESSAGE_FIELD_NAME, message));

        // 检查会话状态并尝试发送错误消息
        if (session.isOpen()) {
            session.sendMessage(new TextMessage(errorMessage));
        } else {
            logger.warn("会话已关闭，无法发送错误消息: {}", message);
        }
    } catch (Exception e) {
        // 捕获所有可能的异常，防止影响主流程
        logger.error("发送错误消息失败 (会话ID: {})", session.getId(), e);
    }
}
```


#### 10.3 任务创建与初始化

`createTask()`方法负责创建和初始化任务实例：

1. **参数提取**：从消息中提取任务名称和参数
2. **类型查找**：根据任务名称查找对应的类名
3. **实例创建**：通过反射创建任务实例
4. **国际化设置**：设置任务的Locale信息
5. **处理器绑定**：为任务绑定各类消息处理器
6. **参数初始化**：调用任务的`init()`方法进行参数初始化
7. **异步执行**：使用虚拟线程启动任务执行

```java
private void createTask(WebSocketSession session, String sessionId, JsonNode jsonNode) {
    if (!jsonNode.has(WebSocketConstant.ARGUMENTS_FIELD_NAME)) {
        throw new IllegalArgumentException("缺失 " + WebSocketConstant.ARGUMENTS_FIELD_NAME + " 参数");
    }
    JsonNode argsNode = jsonNode.get(WebSocketConstant.ARGUMENTS_FIELD_NAME);

    String taskName = argsNode.get(0).asText();
    String clzName = taskClzMap.get(taskName);
    if (clzName == null) {
        throw new IllegalArgumentException("未知的任务类型: " + taskName);
    }
    Object tasker = null;
    try {
        Class<?> clz = Class.forName(clzName);
        tasker = clz.getDeclaredConstructor().newInstance();
    } catch (ClassNotFoundException e) {
        throw new IllegalArgumentException("未知的任务类型: " + taskName + ", class: " + clzName);
    } catch (Exception e) {
        throw new IllegalArgumentException("任务类型: " + taskName + ", class: " + clzName + " 实例化失败");
    }

    if (tasker instanceof Tasker<?> t) {
        String locale = argsNode.get(1).asText();
        t.setLocale(Locale.forLanguageTag(locale));
    }

    if (tasker instanceof WebSocketServerTasker t) {
        t.setTitleHandler(title -> sendToSession(session, sessionId, "title", title));
        t.setMessageHandler(msg -> sendMessageToSession(session, sessionId, msg));
        t.setPropertyHandler((name, value) -> sendToSession(session, sessionId, "property", name, value));
        t.setProgressHandler((current, total) -> sendToSession(session, sessionId, "progress", current, total));
        t.init(argsNode.get(2));
    }

    if (tasker instanceof Callable<?> callable) {
        Thread.ofVirtual().start(() -> {
            try {
                sendToSession(session, sessionId, "start");
                callable.call();
                sendToSession(session, sessionId, "done");
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        });
    }
}
```

#### 10.4 消息通信机制

`WebSocketServerTaskManager`提供了多种消息发送方法，实现服务器端到客户端的通信：

- **`sendToSession()`**：发送各类消息（title、property、progress、start、done等）
- **`sendMessageToSession()`**：专门发送带级别的消息
- **`sendError()`**：发送错误消息给客户端

```java
private boolean sendToSession(WebSocketSession session, String sessionId, String type, Object... args) {
    try {
        String text = objectMapper.writeValueAsString(Map.of(
                WebSocketConstant.SESSION_ID_FIELD_NAME, sessionId,
                "type", type,
                WebSocketConstant.ARGUMENTS_FIELD_NAME, args));
        session.sendMessage(new TextMessage(text));
    } catch (IOException e) {
        // 捕获所有可能的异常，防止影响主流程
        logger.error("发送错误消息失败 (会话ID: {})", session.getId(), e);
    }
    return true;
}

private boolean sendMessageToSession(WebSocketSession session, String sessionId, Message msg) {
    return sendToSession(session, sessionId, "message", msg.getLevel().getName(), msg.getMessage());
}

private void sendError(WebSocketSession session, String sessionId, String message) {
    if (session == null || !session.isOpen()) {
        logger.warn("尝试向已关闭的WebSocket会话发送错误消息: {}", message);
        return;
    }

    try {
        String errorMessage = objectMapper.writeValueAsString(Map.of(
                WebSocketConstant.SESSION_ID_FIELD_NAME, sessionId,
                WebSocketConstant.SUCCESS_FIELD_VALUE, false,
                WebSocketConstant.MESSAGE_FIELD_NAME, message));

        // 检查会话状态并尝试发送错误消息
        if (session.isOpen()) {
            session.sendMessage(new TextMessage(errorMessage));
        } else {
            logger.warn("会话已关闭，无法发送错误消息: {}", message);
        }
    } catch (Exception e) {
        // 捕获所有可能的异常，防止影响主流程
        logger.error("发送错误消息失败 (会话ID: {})", session.getId(), e);
    }
}

#### 10.5 与WebSocketServerHandler的关系

`WebSocketServerTaskManager`与`WebSocketServerHandler`协同工作，共同处理客户端的WebSocket请求：

- `WebSocketServerHandler`作为WebSocket消息的入口点，接收所有客户端消息
- 对于与任务相关的消息（如创建任务），`WebSocketServerHandler`将消息转发给`WebSocketServerTaskManager`处理
- `WebSocketServerTaskManager`负责具体的任务创建、初始化和执行逻辑
- 两者结合实现了完整的客户端-服务器端Tasker通信机制

## 11. WebSocketClientService工作机制

### 11.1 连接管理流程

`WebSocketClientService`的连接管理流程包括以下几个核心步骤：

1. **初始化连接**：通过`initWebSocket()`方法创建与服务器的WebSocket连接
2. **连接状态监控**：使用`online`和`message`属性实时反映连接状态
3. **自动重连机制**：当连接断开时，通过`scheduleReconnect()`方法安排自动重连
4. **心跳维护**：通过`startHeartbeat()`和`heartbeat()`方法定期发送ping消息

### 11.2 消息处理机制

`WebSocketClientService`使用内置的`WebSocketListener`处理各类WebSocket事件：

- `onOpen`：连接建立时触发，启动心跳并更新连接状态
- `onMessage`：接收服务器消息时触发，解析并路由消息到对应处理器
- `onClosing`/`onClosed`：连接关闭时触发，停止心跳并准备重连
- `onFailure`：连接失败时触发，记录错误并尝试重连

### 11.3 消息路由策略

`WebSocketClientService`采用以下策略路由接收到的消息：

1. **回调消息**：如果消息包含`messageId`，则路由到对应的`CompletableFuture`回调
2. **会话消息**：如果消息包含`sessionId`，则路由到对应的`WebSocketClientSession`处理
3. **错误消息**：如果消息包含错误码，则按错误类型进行处理（如未授权时自动重新登录）

## 12. WebSocketClientSession工作机制

### 12.1 会话生命周期管理

每个`WebSocketClientSession`实例的生命周期包括：

1. **创建**：通过`WebSocketClientService.createSession()`方法创建
2. **任务提交**：通过`submitTask()`方法提交Tasker任务到服务器
3. **消息处理**：处理服务器返回的各类消息
4. **关闭**：任务完成后通过`close()`方法关闭会话

### 12.2 消息类型处理

`WebSocketClientSession`能够处理以下几种主要的消息类型：

- **message**：普通文本消息，通过`handleAsMessage()`处理
- **title**：任务标题更新，通过`handleAsTitle()`处理
- **property**：任务属性更新，通过`handleAsProperty()`处理
- **progress**：任务进度更新，通过`handleAsProgress()`处理
- **start**：任务开始通知，通过`handleAsStart()`处理
- **done**：任务完成通知，通过`handleAsDone()`处理

### 12.3 属性同步机制

`WebSocketClientSession`的`handleAsProperty()`方法实现了服务器到客户端的属性同步机制：

1. 接收属性名和属性值
2. 通过Java反射获取Tasker类的属性描述符
3. 将接收到的属性值转换为正确的Java类型
4. 调用属性的setter方法更新Tasker实例的属性值

## 13. 最佳实践建议

1. **任务拆分**：复杂任务应拆分为多个小任务，便于进度跟踪和错误定位
2. **状态反馈**：在关键节点提供清晰的状态信息，增强用户体验
3. **资源释放**：确保文件流等资源正确关闭，避免资源泄露
4. **事务控制**：对于涉及多步数据操作的任务，考虑使用事务确保数据一致性
5. **错误重试**：针对网络波动等临时性问题，考虑实现任务重试机制
6. **配置管理**：使用`tasker_mapper.json`文件统一管理任务注册信息，避免在代码中硬编码任务映射
7. **异常处理**：确保实现配置文件加载失败的fallback机制，保证系统稳定性
8. **版本控制**：对任务配置文件进行版本控制，便于追踪变更历史
9. **会话管理**：确保任务完成后正确关闭WebSocket会话，避免资源泄露
10. **属性同步**：合理使用属性同步机制，确保客户端显示与服务器状态一致

通过遵循以上规则和模式，可以确保Contract-Manager项目中客户端与服务器端Tasker通信的一致性、可靠性和可维护性。